欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

spring集成ActiveMQ, 消息提供者

程序员文章站 2022-06-07 19:14:47
...
1.spring集成ActiveMQ,依赖包
<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-jms</artifactId>
	<version>4.2.6.RELEASE</version>
</dependency>
<dependency>
	<groupId>org.apache.activemq</groupId>
	<artifactId>activemq-client</artifactId>
	<version>5.13.3</version>
</dependency>
<dependency>
	<groupId>org.apache.activemq</groupId>
	<artifactId>activemq-spring</artifactId>
	<version>5.13.3</version>
</dependency>

2.配置spring文件,主要配置ConnectionFactory,和ActiveMQDestination,JSMTemplate
<!-- ActiveMQConnectionFactory -->
<bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
	<constructor-arg index="0" value="tcp://127.0.0.1:61616"/>
	<property name="trustedPackages">
		<list>
			<value>com.lance.mq</value>
			<value>java.util</value>
		</list>
	</property>
</bean>

<!-- 创建pooledConnectionFactory连接池, JmsTemlate每次发送消息都重新打开一个Connection -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
	<property name="connectionFactory" ref="activeMQConnectionFactory"/>
	<property name="maxConnections" value="5"/>
</bean>


<!-- CachingConnectionFactory 
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
	<constructor-arg index="0" ref="activeMQConnectionFactory"/>
</bean>
-->

<!-- jmsTemplate -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
	 <property name="connectionFactory" ref="pooledConnectionFactory"/>
	 <property name="messageConverter">
		<bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
	 </property>
</bean>

<!-- ActiveMQQueue -->
<bean id="activeMQQueue" class="org.apache.activemq.command.ActiveMQQueue">
	<constructor-arg index="0" value="queue"/>
</bean>

3.消息发送
/**
 * 发送Text消息队列
 * @param message
 */
public void sendTextQueueMessage(final String message) {
	jmsTemplate.send(activeMQQueue, new MessageCreator(){
		@Override
		public Message createMessage(Session session) throws JMSException {
			return session.createTextMessage(message);
		}
	});
}

/**
 * 发送消息并处理消息返回值
 * @param message
 * @throws JMSException 
 */
public String sendTextQueueMessageAndReceive(final String message) throws JMSException {
	Message replyMessage = jmsTemplate.sendAndReceive(activeMQQueue, new MessageCreator(){
		@Override
		public Message createMessage(Session session) throws JMSException {
			return session.createTextMessage(message);
		}
	});
	
	log.info("sendTextAndReceive: {}", JSON.toJSONString(replyMessage));
	TextMessage textMessage = (TextMessage)replyMessage;
	return textMessage.getText();
}

/**
 * 订单入队列
 * @param info
 * @throws JMSException 
 */
public String sendObjectQueueMessage(final OrderInfo info) throws JMSException {
	Message message = jmsTemplate.sendAndReceive(activeMQQueue, new MessageCreator() {
		@Override
		public Message createMessage(Session session) throws JMSException {
			return session.createObjectMessage(info);
		}
	});
	
	log.info("sendTextAndReceive: {}", JSON.toJSONString(message));
	TextMessage textMessage = (TextMessage)message;
	return textMessage.getText();
}


5.可以模拟多个生产者,在高并发下,多台服务器请求进入消息队列,交给消费者统一生成订单,详见demo地址:
https://github.com/leelance/demo/tree/master/demo-springmvc-activemq