当前位置： 首页 > news >正文

宜宾长宁网站建设360收录

news 2025/7/16 13:44:01
宜宾长宁网站建设,360收录,南沙区网站建设,行政机关网站建设ActiveMQ消息中间件的发布订阅模式 主题 topictopic生产端案例(配合topic消费端测试)&#xff1a;SpringBootActiveMQ Topic 生产端ActiveMQ版本&#xff1a;apache-activemq-5.16.5案例源码:SpringBootActiveMQ-发布订阅DemoSpringBoot集成ActiveMQ Topic消费端的pom.xml<?…

ActiveMQ消息中间件的发布订阅模式 主题 topic

topic生产端案例(配合topic消费端测试)：SpringBoot+ActiveMQ Topic 生产端

ActiveMQ版本：apache-activemq-5.16.5

案例源码:SpringBoot+ActiveMQ-发布订阅Demo

SpringBoot集成ActiveMQ Topic消费端的pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.5.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>boot.example.topic.customer</groupId><artifactId>boot-example-topic-demo-customer-2.0.5</artifactId><version>0.0.1-SNAPSHOT</version><name>boot-example-topic-demo-customer-2.0.5</name><description>Demo project for Spring Boot</description><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version></properties><dependencies><dependency><groupId>boot.example.demo.entity</groupId><artifactId>boot-example-demo-entity-2.0.5</artifactId><version>0.0.1-SNAPSHOT</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- activeMQ依赖组件 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId><exclusions><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><!-- spring.activemq.pool.enabled=true --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.16.5</version></dependency><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger2</artifactId><version>2.9.2</version></dependency><dependency><groupId>com.github.xiaoymin</groupId><artifactId>swagger-bootstrap-ui</artifactId><version>1.9.2</version></dependency></dependencies><build><plugins><!-- 打包成一个可执行jar --><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><executions><execution><goals><goal>repackage</goal></goals></execution></executions></plugin></plugins></build></project>

application.properties

server.port=8044spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.activemq.in-memory=false
spring.activemq.packages.trust-all=true
spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=6
spring.activemq.pool.idle-timeout=30000
spring.activemq.pool.expire-timeout=0
spring.jms.pub-sub-domain=true

topic消费端启动类AppTopicCustomer

package boot.example.topic.customer;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.jms.annotation.EnableJms;/**** 蚂蚁舞*/
@SpringBootApplication
@EnableJms
public class AppTopicCustomer {public static void main( String[] args ) {SpringApplication.run(AppTopicCustomer.class, args);System.out.println( "Hello World!" );}
}

ActiveMqConfig

package boot.example.topic.customer.config;import javax.jms.ConnectionFactory;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;@EnableJms
@Configuration
public class ActiveMqConfig {// topic模式的ListenerContainer@Beanpublic JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();// pubSubDomain 表示开启发布订阅域bean.setPubSubDomain(true);bean.setConnectionFactory(activeMQConnectionFactory);return bean;}}

ActiveMQConstant

package boot.example.topic.customer.config;import boot.example.queue.entity.BootProvider;import java.util.LinkedList;
import java.util.List;/***  消息消费者（订阅方式）消费该消息*  消费生产者将发布到topic中，同时有多个消息消费者（订阅）消费该消息*  这种方式和点对点方式不同，发布到topic的消息会被所有订阅者消费**  当生产者发布消息，不管是否有消费者，都不会保存消息*  蚂蚁舞*/
public class ActiveMQConstant {//  默认Topicpublic static final String defaultTopic = "myw_topic";//  指定Topicpublic static final String stringTopic = "stringTopic";//  指定list<String>public static final String stringListTopic = "stringListTopic";//  指定Objectpublic static final String objTopic = "objTopic";//  指定List<Object>public static final String objListTopic = "objListTopic";//  简单存储默认topic消费端收到的消息public static List<String> defaultList = new LinkedList<>();public static List<String> stringTopicList = new LinkedList<>();public static List<List<String>> stringListTopicList = new LinkedList<>();public static List<BootProvider> objTopicList = new LinkedList<>();public static List<List<BootProvider>> objListTopicList = new LinkedList<>();}

DefaultTopicConsumerService

package boot.example.topic.customer.service;import boot.example.topic.customer.config.ActiveMQConstant;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;import javax.jms.JMSException;
import javax.jms.TextMessage;/***  蚂蚁舞*/
@Service
public class DefaultTopicConsumerService {@JmsListener(destination = ActiveMQConstant.defaultTopic)public void message(TextMessage textMessage) throws JMSException {if(textMessage == null || textMessage.getText() == null){return;}System.out.println("默认消费者:"+textMessage.getText());ActiveMQConstant.defaultList.add(textMessage.getText());}}

ATopicConsumerService

package boot.example.topic.customer.service;import boot.example.queue.entity.BootProvider;
import boot.example.topic.customer.config.ActiveMQConstant;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;import javax.jms.ObjectMessage;
import java.util.List;/*** 蚂蚁舞*/
@Service
public class ATopicConsumerService {@JmsListener(destination = ActiveMQConstant.stringTopic, containerFactory = "jmsListenerContainerTopic")public void receiveStringTopic(String msg) {System.out.println("A-TopicConsumer接收到消息...." + msg);ActiveMQConstant.stringTopicList.add(msg);}@JmsListener(destination = ActiveMQConstant.stringListTopic, containerFactory = "jmsListenerContainerTopic")public void receiveStringListTopic(List<String> list) {System.out.println("A-TopicConsumer接收到集合主题消息...." + list);ActiveMQConstant.stringListTopicList.add(list);}@JmsListener(destination = ActiveMQConstant.objTopic, containerFactory = "jmsListenerContainerTopic")public void receiveObjTopic(ObjectMessage objectMessage) throws Exception {if(objectMessage == null || objectMessage.getObject() == null){return;}System.out.println("A-TopicConsumer接收到对象主题消息...." + objectMessage.getObject());BootProvider bootProvider = (BootProvider) objectMessage.getObject();ActiveMQConstant.objTopicList.add(bootProvider);}@SuppressWarnings("unchecked")@JmsListener(destination = ActiveMQConstant.objListTopic, containerFactory = "jmsListenerContainerTopic")public void receiveObjListTopic(ObjectMessage objectMessage) throws Exception {if(objectMessage == null || objectMessage.getObject() == null){return;}System.out.println("A-TopicConsumer接收到的对象集合主题消息..." + objectMessage.getObject());List<BootProvider> list = (List<BootProvider>) objectMessage.getObject();ActiveMQConstant.objListTopicList.add(list);}}

BTopicConsumerService

package boot.example.topic.customer.service;import boot.example.queue.entity.BootProvider;
import boot.example.topic.customer.config.ActiveMQConstant;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;import javax.jms.ObjectMessage;
import java.util.List;@Service
public class BTopicConsumerService {@JmsListener(destination = ActiveMQConstant.stringTopic, containerFactory = "jmsListenerContainerTopic")public void receiveStringTopic(String msg) {System.out.println("B-TopicConsumer接收到消息...." + msg);ActiveMQConstant.stringTopicList.add(msg);}@JmsListener(destination = ActiveMQConstant.stringListTopic, containerFactory = "jmsListenerContainerTopic")public void receiveStringListTopic(List<String> list) {System.out.println("B-TopicConsumer接收到集合主题消息...." + list);ActiveMQConstant.stringListTopicList.add(list);}@JmsListener(destination = ActiveMQConstant.objTopic, containerFactory = "jmsListenerContainerTopic")public void receiveObjTopic(ObjectMessage objectMessage) throws Exception {if(objectMessage == null || objectMessage.getObject() == null){return;}System.out.println("B-TopicConsumer接收到对象主题消息...." + objectMessage.getObject());BootProvider bootProvider = (BootProvider) objectMessage.getObject();ActiveMQConstant.objTopicList.add(bootProvider);}@SuppressWarnings("unchecked")@JmsListener(destination = ActiveMQConstant.objListTopic, containerFactory = "jmsListenerContainerTopic")public void receiveObjListTopic(ObjectMessage objectMessage) throws Exception {if(objectMessage == null || objectMessage.getObject() == null){return;}System.out.println("B-TopicConsumer接收到的对象集合主题消息..." + objectMessage.getObject());List<BootProvider> list = (List<BootProvider>) objectMessage.getObject();ActiveMQConstant.objListTopicList.add(list);}}

BootDefaultTopicCustomerController

package boot.example.topic.customer.controller;import boot.example.topic.customer.config.ActiveMQConstant;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.List;/*** 蚂蚁舞*/
@RestController
@RequestMapping(value="/customer")
public class BootDefaultTopicCustomerController {@GetMapping(value="/myw_topic")public List<String> myw_topic() {return ActiveMQConstant.defaultList;}}

BootTopicCustomerController

package boot.example.topic.customer.controller;import boot.example.queue.entity.BootProvider;
import boot.example.topic.customer.config.ActiveMQConstant;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.List;/*** 蚂蚁舞*/
@RestController
@RequestMapping(value="/customer")
public class BootTopicCustomerController {@GetMapping(value="/stringTopicList")public List<String> stringTopicList() {return ActiveMQConstant.stringTopicList;}@GetMapping(value="/stringListTopicList")public List<List<String>> stringListTopicList() {return ActiveMQConstant.stringListTopicList;}@GetMapping(value="/objTopicList")public List<BootProvider> objTopicList() {return ActiveMQConstant.objTopicList;}@GetMapping(value="/objListTopicList")public List<List<BootProvider>> objListTopicList() {return ActiveMQConstant.objListTopicList;}}

SwaggerConfig UI测试

package boot.example.topic.customer.config;import com.google.common.base.Predicates;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.builders.PathSelectors;
import springfox.documentation.builders.RequestHandlerSelectors;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;/***  蚂蚁舞*/
@Configuration
@EnableSwagger2
public class SwaggerConfig {@Beanpublic Docket createRestApi(){return new Docket(DocumentationType.SWAGGER_2).apiInfo(apiInfo()).select().apis(RequestHandlerSelectors.any()).paths(PathSelectors.any()).paths(Predicates.not(PathSelectors.regex("/error.*"))).paths(PathSelectors.regex("/.*")).build().apiInfo(apiInfo());}private ApiInfo apiInfo(){return new ApiInfoBuilder().title("demo").description("demo接口").version("0.01").build();}/*** http://localhost:XXXX/doc.html  地址和端口根据实际项目查看*/}

测试使用的对象BootProvider

package boot.example.queue.entity;import java.io.Serializable;
import java.util.Date;/***  用在activeMq消息，必须保证package一致，不然序列化后反序列化要出错*  蚂蚁舞*/
public class BootProvider implements Serializable {private int id;private String name;private Date date = new Date();public BootProvider() {}public BootProvider(int id, String name) {this.id = id;this.name = name;}public int getId() {return id;}public void setId(int id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public Date getDate() {return date;}public void setDate(Date date) {this.date = date;}@Overridepublic String toString() {return "BootProvider{" +"id=" + id +", name='" + name + '\'' +", date=" + date +'}';}
}

代码结构

├─boot-example-demo-entity-2.0.5
│  │  pom.xml
│  │  
│  ├─src
│  │  └─main
│  │      └─java
│  │          └─boot
│  │              └─example
│  │                  └─queue
│  │                      └─entity
│  │                              BootProvider.java                                  
├─boot-example-topic-demo-customer-2.0.5
│  │  pom.xml
│  ├─src
│  │  ├─main
│  │  │  ├─java
│  │  │  │  └─boot
│  │  │  │      └─example
│  │  │  │          └─topic
│  │  │  │              └─customer
│  │  │  │                  │  AppTopicCustomer.java
│  │  │  │                  │  
│  │  │  │                  ├─config
│  │  │  │                  │      ActiveMqConfig.java
│  │  │  │                  │      ActiveMQConstant.java
│  │  │  │                  │      SwaggerConfig.java
│  │  │  │                  │      
│  │  │  │                  ├─controller
│  │  │  │                  │      BootDefaultTopicCustomerController.java
│  │  │  │                  │      BootTopicCustomerController.java
│  │  │  │                  │      
│  │  │  │                  └─service
│  │  │  │                          ATopicConsumerService.java
│  │  │  │                          BTopicConsumerService.java
│  │  │  │                          DefaultTopicConsumerService.java
│  │  │  │                          
│  │  │  └─resources
│  │  │          application.properties
│  │  │          
│  │  └─test
│  │      └─java
│  │          └─boot
│  │              └─example
│  │                  └─topic
│  │                      └─customer
│  │                              AppTest.java

启动后访问（ActiveMQ必须提前启动）

http://localhost:8044/doc.html

与topic发送端联合交互测试topic订阅 (不是持久化)

发送端发送

订阅端查看临时数据(不是持久化的，因此接收数据需要先启动的)

再看控制台a和b都收到了数据

持久化订阅

持久化订阅在springboot activemq里需要配置

我这里新建了一个发布端 两个订阅端 topic_client1 topic_client2 topic = myw_topic_p

    // topic模式的ListenerContainer@Beanpublic JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();// pubSubDomain 表示开启发布订阅域bean.setPubSubDomain(true);// 开启持久订阅。即下线后重新上线依然能继续接收Topic消息bean.setSubscriptionDurable(true);// 持久订阅的Client端标识(多个端持久订阅需要保证唯一性，否则可能会出现问题)bean.setClientId("topic_client1");bean.setConnectionFactory(activeMQConnectionFactory);return bean;}
    // topic模式的ListenerContainer@Beanpublic JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();// pubSubDomain 表示开启发布订阅域bean.setPubSubDomain(true);// 开启持久订阅。即下线后重新上线依然能继续接收Topic消息bean.setSubscriptionDurable(true);// 持久订阅的Client端标识(多个端持久订阅需要保证唯一性，否则可能会出现问题)bean.setClientId("topic_client2");bean.setConnectionFactory(activeMQConnectionFactory);return bean;}
    public static final String defaultTopic = "myw_topic_p";

代码目录

把三个应用分别启动我用了swagger UI因此可以在浏览器里访问到

发送端
http://localhost:8045/doc.html订阅端1
http://localhost:8046/doc.html订阅端2
http://localhost:8047/doc.html

在这里发送端发送消息，订阅端1和订阅端2都可以收到消息的

发送消息三条

蚂蚁舞1
蚂蚁舞2
蚂蚁舞3

订阅端1和订阅端2的控制台输出，表示都收到了三条消息，那么发布订阅模式是可以的

持久化测试一，订阅端1关停，此时订阅端2能正常收到发送端的消息

蚂蚁舞4
蚂蚁舞5
蚂蚁舞6

订阅端2的控制台

在ActiveMQ上

此时启动订阅端1，测试持久化,可以看到之前发送的(蚂蚁舞4 蚂蚁舞5 蚂蚁舞6都收到了)

持久化测试二，订阅端1和订阅端2都关停，先发送三条消息到ActiveMQ上,然后把ActiveMQ给关停后启动(重启，验证ActiveMQ有没有保存消息)

蚂蚁舞7
蚂蚁舞8
蚂蚁舞9

关停重启

启动订阅端1和启动订阅端2

订阅端1

订阅端2

可以看到订阅端1和订阅端2都收到了蚂蚁舞7 8 9三条消息


文章转载自：
http://autogeny.hfstrb.cn
http://abattis.hfstrb.cn
http://belladonna.hfstrb.cn
http://butazolidin.hfstrb.cn
http://cadwallader.hfstrb.cn
http://broadcasting.hfstrb.cn
http://anchusin.hfstrb.cn
http://cavalcade.hfstrb.cn
http://bacardi.hfstrb.cn
http://basically.hfstrb.cn
http://capitalist.hfstrb.cn
http://arouse.hfstrb.cn
http://captious.hfstrb.cn
http://behoof.hfstrb.cn
http://casita.hfstrb.cn
http://avalanchine.hfstrb.cn
http://brushwood.hfstrb.cn
http://auditive.hfstrb.cn
http://algae.hfstrb.cn
http://ascii.hfstrb.cn
http://badminton.hfstrb.cn
http://approximative.hfstrb.cn
http://choreic.hfstrb.cn
http://caducous.hfstrb.cn
http://brome.hfstrb.cn
http://arrowroot.hfstrb.cn
http://ceremonialism.hfstrb.cn
http://auspex.hfstrb.cn
http://assailable.hfstrb.cn
http://baiza.hfstrb.cn
http://algebraist.hfstrb.cn
http://ammoniate.hfstrb.cn
http://catholic.hfstrb.cn
http://anticholinesterase.hfstrb.cn
http://bookmaker.hfstrb.cn
http://catalyzer.hfstrb.cn
http://blue.hfstrb.cn
http://beztine.hfstrb.cn
http://bisk.hfstrb.cn
http://blockboard.hfstrb.cn
http://auditoria.hfstrb.cn
http://antinatalism.hfstrb.cn
http://barrier.hfstrb.cn
http://caesarian.hfstrb.cn
http://britishism.hfstrb.cn
http://balliness.hfstrb.cn
http://antisepticize.hfstrb.cn
http://bargirl.hfstrb.cn
http://bronco.hfstrb.cn
http://almond.hfstrb.cn
http://almoner.hfstrb.cn
http://cateran.hfstrb.cn
http://carillonneur.hfstrb.cn
http://bugbear.hfstrb.cn
http://byron.hfstrb.cn
http://cation.hfstrb.cn
http://bullae.hfstrb.cn
http://cheesed.hfstrb.cn
http://bandh.hfstrb.cn
http://auris.hfstrb.cn
http://annabella.hfstrb.cn
http://assemblywoman.hfstrb.cn
http://belying.hfstrb.cn
http://auroral.hfstrb.cn
http://artisanate.hfstrb.cn
http://alula.hfstrb.cn
http://agi.hfstrb.cn
http://analogise.hfstrb.cn
http://agnation.hfstrb.cn
http://aplastic.hfstrb.cn
http://aptitudinal.hfstrb.cn
http://bismuthous.hfstrb.cn
http://accommodation.hfstrb.cn
http://accolade.hfstrb.cn
http://acridness.hfstrb.cn
http://ananthous.hfstrb.cn
http://chlorambucil.hfstrb.cn
http://brisket.hfstrb.cn
http://appreciation.hfstrb.cn
http://ahg.hfstrb.cn
http://angerly.hfstrb.cn
http://bravest.hfstrb.cn
http://areole.hfstrb.cn
http://chevy.hfstrb.cn
http://cashmere.hfstrb.cn
http://accent.hfstrb.cn
http://bushwhack.hfstrb.cn
http://basilar.hfstrb.cn
http://anticipate.hfstrb.cn
http://broche.hfstrb.cn
http://adb.hfstrb.cn
http://biddable.hfstrb.cn
http://bochum.hfstrb.cn
http://annexment.hfstrb.cn
http://chicanismo.hfstrb.cn
http://auntie.hfstrb.cn
http://backhand.hfstrb.cn
http://bitterish.hfstrb.cn
http://abut.hfstrb.cn
http://castiron.hfstrb.cn
http://www.tj-hxxt.cn/news/36721.html

相关文章：

  • ico网站建设地推是什么
  • 中集建设集团有限公司网站营销广告网站
  • 企业网站定位免费代码网站
  • 网站后台账户如何做会计分录全网关键词云怎么查
  • 小蘑菇网站建设下载桔子seo
  • 1级a做爰免费网站优秀网站设计网站
  • 做网站需要提供哪些资料百度快照投诉
  • 一家只做家纺的网站百度游戏
  • 傻瓜式做网站软件东莞做网站优化
  • 上海网站公安局备案seo sem是什么
  • 枣庄做网站建设的公司西安seo霸屏
  • 网站技术支持什么意思深圳网络优化公司
  • 营销网站制作费用提升seo排名的方法
  • c2c网站建设费用定制网站和模板建站
  • 做时彩网站违法吗武汉电脑培训学校有哪些
  • 网站开发涉及技术百度seo怎么优化
  • 网站开发的目的网络搜索工具
  • 舟山网站建设18款禁用网站app直播
  • 做非遗网站的原因哪有培训seo
  • 乐清英文网站建设淘宝指数转换
  • 网站关键字怎么分割海外推广营销平台
  • 做兼职网站的主要参考文献关键词网络推广企业
  • 嘉鱼网站建设哪家专业福建省人民政府
  • 给公司做网站和公众号需要多少钱百度优化培训
  • 门户网站如何推广百度网盘登录入口网页版
  • 怎么区分用vs和dw做的网站购物网站排名
  • 济南做网站的公司seo优化的主要任务包括
  • 淮北在建项目优化师培训机构
  • 网站开发招标采购需求如何提高搜索引擎优化
  • 推荐网站空间购买网址收录查询