一、新建gradle工程

1.1 使用intellij新建gradle工程步骤省略

二、导入依赖,配置build.gradle

plugins {
id 'java'
} group 'com.bdht'
version '1.0-SNAPSHOT' sourceCompatibility = 1.8 repositories {
mavenCentral()
} dependencies {
testCompile group: 'junit', name: 'junit', version: '4.12'
compile group: 'org.springframework', name: 'spring-context'
compile group: 'org.apache.activemq', name: 'activemq-all', version: '5.15.9'
compile group: 'org.apache.activemq', name: 'activemq-pool', version: '5.15.9'
compile group: 'org.springframework', name: 'spring-jms', version: '5.1.9.RELEASE'
compile group: 'org.springframework', name: 'spring-aop', version: '5.1.9.RELEASE'
}

三、通过注解的方式整合activemq

3.1 在resources目录下配置application.properties

#activemq地址
activemq.url=tcp://192.168.1.66:61616
#队列模式name
activemq.queueName=message-queue
#主题模式name
activemq.topicName=message-topic

3.2 配置生产者,新建producer包,在包下新建ProducerConfig类

package com.bdht.amq.producer;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.connection.SingleConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
@ComponentScan(basePackages={"com.bdht"})
@EnableJms
@Configuration
@PropertySource("classpath:application.properties")
public class ProducerConfig { @Value("${activemq.url}")
private String URL; @Value("${activemq.queueName}")
private String QUEUE_NAME; @Value("${activemq.topicName}")
private String TOPIC_NAME; //配置ConnectionFactory
@Bean
public ActiveMQConnectionFactory activeMQConnectionFactory(){
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(URL);
return activeMQConnectionFactory;
} //配置Connection
@Bean
public SingleConnectionFactory singleConnectionFactory(ActiveMQConnectionFactory activeMQConnectionFactory){
SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory();
singleConnectionFactory.setTargetConnectionFactory(activeMQConnectionFactory);
return singleConnectionFactory;
} //配置jms模板
@Bean
public JmsTemplate jmsTemplate(SingleConnectionFactory singleConnectionFactory){
JmsTemplate jmsTemplate = new JmsTemplate();
jmsTemplate.setConnectionFactory(singleConnectionFactory);
return jmsTemplate;
} //配置队列目的地
@Bean
public ActiveMQQueue activeMQQueue(){
ActiveMQQueue activeMQQueue = new ActiveMQQueue(QUEUE_NAME);
return activeMQQueue;
} //配置主题目的地
@Bean
public ActiveMQTopic activeMQTopic(){
ActiveMQTopic activeMQTopic = new ActiveMQTopic(TOPIC_NAME);
return activeMQTopic;
} }

3.3 配置监听器,监听器配置在consumer包下,分别新建队列模式监听器QueueMsgListener和主题模式监听器TopicMsgListener

package com.bdht.amq.consumer;

import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
@Component
public class QueueMsgListener implements MessageListener {
@Override
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage)message;
System.out.println("queue监听者正在监听...");
try {
System.out.println("queue监听者收到消息:"+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
package com.bdht.amq.consumer;

import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
@Component
public class TopicMsgListener implements MessageListener {
@Override
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
System.out.println("topic监听者正在监听消息...");
try {
System.out.println("topic监听者监听到消息"+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
} }
}

3.4 配置消费者,在consumer包下新建ConSumerConfig类

package com.bdht.amq.consumer;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.connection.SingleConnectionFactory;
import org.springframework.jms.listener.DefaultMessageListenerContainer; @ComponentScan(basePackages={"com.bdht"})
@EnableJms
@Configuration
@PropertySource("classpath:application.properties")
public class ConsumerConfig { @Value("${activemq.url}")
private String URL; @Value("${activemq.queuqName}")
private String QUEUE_NAME; @Value("${activemq.topicName}")
private String TOPIC_NAME; //配置ConnectionFactory
@Bean
public ActiveMQConnectionFactory activeMQConnectionFactory(){
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(URL);
return activeMQConnectionFactory;
} //配置连接池
@Bean
public SingleConnectionFactory singleConnectionFactory(ActiveMQConnectionFactory activeMQConnectionFactory){
SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory();
singleConnectionFactory.setTargetConnectionFactory(activeMQConnectionFactory);
return singleConnectionFactory;
} //配置topic监听容器
@Bean
public DefaultMessageListenerContainer defaultMessageListenerContainer(SingleConnectionFactory singleConnectionFactory, TopicMsgListener topicMsgListener,ActiveMQTopic activeMQTopic){
//创建容器
DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
//设置监听器
defaultMessageListenerContainer.setMessageListener(topicMsgListener);
//设置连接
defaultMessageListenerContainer.setConnectionFactory(singleConnectionFactory);
//设置监听目的地
defaultMessageListenerContainer.setDestination(activeMQTopic);
return defaultMessageListenerContainer;
} //配置queue监听容器
@Bean
public DefaultMessageListenerContainer defaultMessageListenerContainerQueue(SingleConnectionFactory singleConnectionFactory, QueueMsgListener queueMsgListener,ActiveMQQueue activeMQQueue){
//创建容器
DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
//设置监听器
defaultMessageListenerContainer.setMessageListener(queueMsgListener);
//设置连接工厂
defaultMessageListenerContainer.setConnectionFactory(singleConnectionFactory);
//设置监听目的地
defaultMessageListenerContainer.setDestination(activeMQQueue);
return defaultMessageListenerContainer;
} //配置队列目的地
@Bean
public ActiveMQQueue activeMQQueue(){
ActiveMQQueue activeMQQueue = new ActiveMQQueue(QUEUE_NAME);
return activeMQQueue;
} //配置主题目的地
@Bean
public ActiveMQTopic activeMQTopic(){
ActiveMQTopic activeMQTopic = new ActiveMQTopic(TOPIC_NAME);
return activeMQTopic;
}
}

3.5 测试,新建demoTest测试类

package com.bdht.amq.test;

import com.bdht.amq.producer.ProducerConfig;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator; import javax.jms.*; public class DemoTest {
public static void main(String[] args) {
//测试主题模式
AnnotationConfigApplicationContext aContext = new AnnotationConfigApplicationContext(ProducerConfig.class);
JmsTemplate jmsTemplate = aContext.getBean(JmsTemplate.class);
Destination bean = (Destination) aContext.getBean("activeMQTopic");
jmsTemplate.send(bean, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage();
textMessage.setText("tttttt......");
return textMessage;
}
});
//测试队列模式
AnnotationConfigApplicationContext aContextQueue = new AnnotationConfigApplicationContext(ProducerConfig.class);
JmsTemplate jmsTemplateQueue = aContextQueue.getBean(JmsTemplate.class);
Destination beanQueue = (Destination) aContextQueue.getBean("activeMQQueue");
jmsTemplateQueue.send(beanQueue, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage();
textMessage.setText("qqqqqq......");
return textMessage;
}
});
}
}

四、通过xml配置的方式整合activemq

4.1 配置application.xml

<?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:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.0.xsd">
<!--生产者和消费者都需要用到的-start-->
<!--配置包扫描路径-->
<context:component-scan base-package="com.bdht.jms.consumer"/>
<context:component-scan base-package="com.bdht.jms.producer"/> <!--ActiveMQ为我们提供的ConnectionFactory-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.1.66:61616"/>
</bean>
<!--spring jms 为我们提供的连接池,这个connectionFactory也是下面的jmsTemplate要使用的,
它其实就是对activeMQ的ConnectionFactory的一个封装-->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean> <!--提供一个队列模式的目的地,点对点的-->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<!--目的地队列的名字-->
<constructor-arg value="queue-demo"/>
</bean>
<!--生产者和消费者都需要用到的-end--> <!--生产者所需要的-start-->
<!--配置jmsTemplate用于发送消息-->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
<bean id="producer" class="com.bdht.jms.producer.Producer"></bean>
<!--生产者所需要的-end--> <!--消费者所需要的start-->
<!--配置消息监听器1,即消息的消费者-->
<bean id="queueListener1" class="com.bdht.jms.consumer.QueueListener1"/>
<!--配置消息容器1-->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<!--指定连接工厂-->
<property name="connectionFactory" ref="connectionFactory"/>
<!--指定消息目的地-->
<property name="destination" ref="queueDestination"/>
<!--指定消息监听器-->
<property name="messageListener" ref="queueListener1"/>
</bean>
<!--&lt;!&ndash;配置消息监听器2,即消息的消费者&ndash;&gt;-->
<!--<bean id="queueListener2" class="com.caihao.activemqdemo.consumer.QueueListener2"/>-->
<!--&lt;!&ndash;配置消息容器2&ndash;&gt;-->
<!--<bean id="jmsContainer2" class="org.springframework.jms.listener.DefaultMessageListenerContainer">-->
<!--&lt;!&ndash;指定连接工厂&ndash;&gt;-->
<!--<property name="connectionFactory" ref="connectionFactory"/>-->
<!--&lt;!&ndash;指定消息目的地&ndash;&gt;-->
<!--<property name="destination" ref="queueDestination"/>-->
<!--&lt;!&ndash;指定消息监听器&ndash;&gt;-->
<!--<property name="messageListener" ref="queueListener2"/>-->
<!--</bean>-->
<!--消费者所需要的end--> </beans>

4.2 配置生产者

package com.bdht.jms.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator; import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session; public class Producer {
@Autowired
private JmsTemplate jmsTemplate;
@Resource(name = "queueDestination")
private Destination queueDestination; public void sendQueueMessage(String message){
jmsTemplate.send(queueDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
}

4.3 配置监听器

package com.bdht.jms.consumer;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage; public class QueueListener1 implements MessageListener { @Override
public void onMessage(Message message) {
if(message instanceof TextMessage){
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("队列模式消费者1:"+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}

4.4 测试

package com.bdht.jms.test;

import com.bdht.jms.producer.Producer;
import org.springframework.context.support.ClassPathXmlApplicationContext; public class Application {
public static void main(String[] args) {
ClassPathXmlApplicationContext ac = new ClassPathXmlApplicationContext("application.xml");
Producer producer = (Producer) ac.getBean("producer");
for(int i=0;i<100;i++){
System.out.println("队列模式生产者:消息"+i);
producer.sendQueueMessage("消息"+i); }
}
}

五、参考优秀博客地址

https://blog.csdn.net/weixin_43732955/article/details/91348042

https://blog.csdn.net/chinese_cai/article/details/100056150

05-11 22:10