模块模式
p2p模式
生产者
ConfigBeanQueue
package com.producerp2p.producerp2p;import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;import javax.jms.Queue;// 配置类--配置文件（Spring的配置文件）
@Configuration
@EnableJms
public class ConfigBeanQueue { // 属性注入的方式
@Value("${queue-boot-name}")
private String queueName; // 创建队列Queue并设置队列名称
@Bean
public Queue createQueue(){
return new ActiveMQQueue(queueName);
}
}
QueueProducer
package com.producerp2p.producerp2p;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;import javax.jms.Queue;@Component
public class QueueProducer { // 属性注入
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate; @Autowired
private Queue queue; // 发送消息
public void sendMsg(){
jmsMessagingTemplate.convertAndSend(queue,"QueueProducer发送消息到对列");
} // 定时发送消息： 每隔3秒发送一条消息 微服务项目启动的时候自动执行使用@Scheduled修饰的方法
@Scheduled(fixedDelay = 3000l)
public void sendMsgScheduled(){
jmsMessagingTemplate.convertAndSend(queue,"QueueProducer发送定时消息到队列");
System.out.println("-------定时消息发送完成-------");
}}
Producerp2pApplication
package com.producerp2p.producerp2p;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication@EnableScheduling
public class Producerp2pApplication { public static void main(String[] args) {
SpringApplication.run(Producerp2pApplication.class, args);
}}
application.properties
server.port=7777
# 配置activemq连接，url
spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.user=admin
spring.activemq.password=admin
# activemq的模式queue（false），topic(true)
spring.jms.pub-sub-domain=false
# 队列名称
queue-boot-name=newqueue
Producerp2pApplicationTests
package com.producerp2p.producerp2p;import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;@SpringBootTest
class Producerp2pApplicationTests { @Autowired
private QueueProducer queueProducer; @Test
void testSendMsg() {
queueProducer.sendMsg();
System.out.println("------消息发送到队列完成-------");
}}
消费者
application.properties
server.port=7778
# 配置activemq连接，url
spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.user=admin
spring.activemq.password=admin
# activemq的模式queue（false），topic(true)
spring.jms.pub-sub-domain=false
# 队列名称
queue-boot-name=newqueue
QueueConsumer
package com.consumerp2p.consumerp2p;import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;import javax.jms.JMSException;
import javax.jms.TextMessage;@Component
public class QueueConsumer { // 接收消息---设置消费者监听的目的地（队列）， 如果监听到队列中有消息，通过方法的形参表示接收到的消息
// Message：TextMessage
@JmsListener(destination = "${queue-boot-name}")
public void receiveMsg(TextMessage textMessage) throws JMSException {
System.out.println("p2p模式下的消费者1接收到队列中的消息："+textMessage.getText()); }
}
Consumerp2pApplication
Consumerp2pApplication11
package com.consumerp2p.consumerp2p;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class Consumerp2pApplication11 { public static void main(String[] args) {
SpringApplication.run(Consumerp2pApplication11.class, args);
}}
——————
———————————————————————————-
———————-
Topic模式
生产者
ConfigBeanTopic
package com.producertopic.producertopic;import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;import javax.jms.Topic;@Configuration
@EnableJms
public class ConfigBeanTopic {
// 注入主题属性
@Value("${topic-boot-name}")
private String topicName; // 创建主题并设置主题名称
@Bean
public Topic createTopic(){
return new ActiveMQTopic(topicName);
}}
ProducerTopic
package com.producertopic.producertopic;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;import javax.jms.Topic;@Component
public class ProducerTopic { // 属性注入
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate; @Autowired
private Topic topic; // 定时发送消息
@Scheduled(fixedDelay = 2000l)
public void sendMsg(){
jmsMessagingTemplate.convertAndSend(topic,"发送定时消息到主题");
System.out.println("-----主题消息发送完成------");
}}
ProducertopicApplication
package com.producertopic.producertopic;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication
@EnableScheduling
public class ProducertopicApplication { public static void main(String[] args) {
SpringApplication.run(ProducertopicApplication.class, args);
}}
application.properties
server.port=7779
# 配置activemq连接，url
spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.user=admin
spring.activemq.password=admin
# activemq的模式queue（false），topic(true)
spring.jms.pub-sub-domain=true
# 主题名称
topic-boot-name=newtopic
消费者
TopicConsumer
package com.consumertopic.consumertopic;import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;import javax.jms.JMSException;
import javax.jms.TextMessage;@Component
public class TopicConsumer { // 接收消息---设置消费者监听的目的地（队列）， 如果监听到队列中有消息，通过方法的形参表示接收到的消息
// Message：TextMessage
@JmsListener(destination = "${topic-boot-name}")
public void receiveMsg(TextMessage textMessage) throws JMSException {
System.out.println("topic下的消费者接收到主题中的消息:"+textMessage.getText());
}
}
application.properties
server.port=7776
# 配置activemq连接，url
spring.activemq.broker-url=tcp://127.0.0.1:61616
spring.activemq.user=admin
spring.activemq.password=admin
# activemq的模式queue（false），topic(true)
spring.jms.pub-sub-domain=true
# 主题名称
topic-boot-name=newtopic
ConsumertopicApplication
package com.consumertopic.consumertopic;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ConsumertopicApplication { public static void main(String[] args) {
SpringApplication.run(ConsumertopicApplication.class, args);
}}
ConsumertopicApplication2
package com.consumertopic.consumertopic;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ConsumertopicApplication2 { public static void main(String[] args) {
SpringApplication.run(ConsumertopicApplication2.class, args);
}}
public class ProducertopicApplication {
