Spring、ActiveMQ整合——Topic模式
程序员文章站
2022-07-13 23:28:11
...
消息生产者
配置文件
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd">
<!-- 需要创建一个连接工厂,连接ActiveMQ. ActiveMQConnectionFactory. 需要依赖ActiveMQ提供的amq标签 -->
<!-- amq:connectionFactory 是bean标签的子标签, 会在spring容器中创建一个bean对象. 可以为对象命名.
类似: <bean id="" class="ActiveMQConnectionFactory"></bean> -->
<amq:connectionFactory brokerURL="tcp://192.168.37.140:61616" userName="admin" password="admin" id="amqConnectionFactory" />
<!-- 配置池化的ConnectionFactory。 为连接ActiveMQ的connectionFactory提供连接池 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactoryBean">
<property name="connectionFactory" ref="amqConnectionFactory"></property>
<property name="maxConnections" value="10"></property>
</bean>
<!-- spring管理JMS相关代码的时候,必须依赖jms标签库. spring-jms提供的标签库. -->
<!-- 定义Spring-JMS中的连接工厂对象 CachingConnectionFactory - spring框架提供的连接工厂对象.
不能真正的访问MOM容器. 类似一个工厂的代理对象. 需要提供一个真实工厂,实现MOM容器的连接访问. -->
<!-- 配置有缓存的ConnectionFactory,session的缓存大小可定制。 -->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
<property name="sessionCacheSize" value="3"></property>
</bean>
<!-- JmsTemplate配置 -->
<bean id="template" class="org.springframework.jms.core.JmsTemplate">
<!-- 给定连接工厂, 必须是spring创建的连接工厂. -->
<property name="connectionFactory" ref="connectionFactory"></property>
<!--开启订阅-->
<property name="pubSubDomain" value="true"></property>
</bean>
</beans>
service实现类
package com.zxw.service.impl;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import com.zxw.pojo.Users;
import com.zxw.service.UserService;
@Service
public class UserServiceImpl implements UserService {
@Autowired
@Qualifier("template")
private JmsTemplate jmsTemplate;
/**
* topic
* @param topicName
* @param user
*/
@Override
public void addUser(String topicName,final Users user) {
//发送消息
this.jmsTemplate.send(topicName,new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
Message message = session.createObjectMessage(user);
return message;
}
});
}
}
控制层
package com.zxw.web.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import com.zxw.pojo.Users;
import com.zxw.service.UserService;
@Controller
@RequestMapping("/user")
public class UserController {
@Autowired
private UserService userService;
@RequestMapping("/addUser")
public String addUser(Users user){
//topic 队列名在这
this.userService.addUser("test.topic",user);
return "ok";
}
}
消费者
配置文件
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd">
<!-- 需要创建一个连接工厂,连接ActiveMQ. ActiveMQConnectionFactory. 需要依赖ActiveMQ提供的amq标签 -->
<!-- amq:connectionFactory 是bean标签的子标签, 会在spring容器中创建一个bean对象.
可以为对象命名. 类似: <bean id="" class="ActiveMQConnectionFactory"></bean>
-->
<amq:connectionFactory brokerURL="tcp://192.168.37.140:61616" userName="admin" password="admin" id="amqConnectionFactory"/>
<!-- spring管理JMS相关代码的时候,必须依赖jms标签库. spring-jms提供的标签库. -->
<!-- 定义Spring-JMS中的连接工厂对象
CachingConnectionFactory - spring框架提供的连接工厂对象. 不能真正的访问MOM容器.
类似一个工厂的代理对象. 需要提供一个真实工厂,实现MOM容器的连接访问.
-->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
<property name="sessionCacheSize" value="3"></property>
</bean>
<!-- 注册监听器 -->
<!-- 开始注册监听.
需要的参数有:
acknowledge - 消息确认机制
container-type - 容器类型 default|simple
simple:SimpleMessageListenerContainer最简单的消息监听器容器,只能处理固定数量的JMS会话,且不支持事务。
default:DefaultMessageListenerContainer是一个用于异步消息监听器容器 ,且支持事务
destination-type - 目的地类型. 使用队列作为目的地.
connection-factory - 连接工厂, spring-jms使用的连接工厂,必须是spring自主创建的
不能使用三方工具创建的工程. 如: ActiveMQConnectionFactory.
-->
<jms:listener-container acknowledge="auto" container-type="default" destination-type="topic" connection-factory="connectionFactory" >
<!-- 在监听器容器中注册某监听器对象.
destination - 设置目的地命名
ref - 指定监听器对象
-->
<!--topic-->
<jms:listener destination="test.topic" ref="myListener"/>
<jms:listener destination="test.topic" ref="myListener2"/>
</jms:listener-container>
</beans>
消息服务监听器myListener(myListener2完全一样)
package com.zxw.listener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.zxw.pojo.Users;
import com.zxw.service.UserService;
/**
* 消息服务监听器
* @author Administrator
* myListener,myListener2一样
*/
@Component(value="myListener")
public class MyMessageListener implements MessageListener{
@Autowired
private UserService userService;
@Override
public void onMessage(Message message) {
//处理消息
ObjectMessage objMessage = (ObjectMessage)message;
Users user=null;
try {
user = (Users)objMessage.getObject();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
this.userService.showUser(user);
}
}
结果
上一篇: 华为笔试——字符串排序、去重、反转等算法(C语言版)
下一篇: 老司机教你怎样下载电影
推荐阅读
-
浅谈Spring Boot 整合ActiveMQ的过程
-
浅谈Spring Boot 整合ActiveMQ的过程
-
Spring Boot(十三):整合Redis哨兵,集群模式实践
-
JAVAEE——宜立方商城09:Activemq整合spring的应用场景、添加商品同步索引库、商品详情页面动态展示与使用缓存
-
Spring整合ActiveMQ---超详细教程
-
Spring整合ActiveMQ如果XSD文件不对导致Spring加载很慢的问题
-
Spring整合ActiveMQ如果XSD文件不对导致Spring加载很慢的问题
-
Spring、ActiveMQ整合——Topic模式
-
Spring整合JMS(一)——基于ActiveMQ实现
-
Spring 整合 ActiveMq