activeMQ入门+spring boot整合activeMQ

最近想要学习MOM（消息中间件：Message Oriented Middleware），就从比较基础的activeMQ学起，rabbitMQ、zeroMQ、rocketMQ、Kafka等后续再去学习。

上面说activeMQ是一种消息中间件，可是为什么要使用activeMQ？

在没有使用JMS的时候，很多应用会出现同步通信（客户端发起请求后需要等待服务端返回结果才能继续执行）、客户端服务端耦合、单一点对点（P2P）通信的问题，JMS可以通过面向消息中间件的方式很好的解决了上面的问题。

JMS规范术语：

Provider/MessageProvider：生产者

Consumer/MessageConsumer：消费者

消息形式： 
1、点对点（queue） 
2、一对多（topic）

ConnectionFactory：连接工厂，JMS用它创建连接

Connnection：JMS Client到JMS Provider的连接

Destination：消息目的地，由Session创建

Session：会话，由Connection创建，实质上就是发送、接受消息的一个线程，因此生产者、消费者都是Session创建的

我这里安装的是Windows版本的，安装好了之后就是这样的目录

到bin目录下，启动activemq.bat

这样就启动成功了。

访问http://localhost:8161/admin/index.jsp可以看到管控台，如下图：

spring boot整合activeMQ：

pom.xml中引用

1<dependency>
2            <groupId>org.springframework.boot</groupId>
3            <artifactId>spring-boot-starter-activemq</artifactId>
4        </dependency>
5        <dependency>
6            <groupId>org.apache.activemq</groupId>
7            <artifactId>activemq-pool</artifactId>
8            <!-- <version>5.7.0</version> -->
9        </dependency>

在application.properties中配置activeMQ连接：

1spring.activemq.broker-url=tcp://localhost:61616
2spring.activemq.in-memory=true
3spring.activemq.pool.enabled=false

创建消息生产者：

1/**
2 * @author huangzhang
3 * @description
4 * @date Created in 2018/7/16 18:57
5 */
6@Service("provider")
7public class Provider {
8    @Autowired
9    private JmsMessagingTemplate jmsMessagingTemplate;
10
11    public void sendMessage(Destination destination, final String message){
12        jmsMessagingTemplate.convertAndSend(destination,message);
13
14    }　　//消费消费者返回的队列"return.queue"中的消息
15    @JmsListener(destination="return.queue")
16    public void consumerMessage(String string){
17        System.out.println("从return.queue队列收到的回复报文为:"+string);
18    }
19
20}

创建第一个消费者：

1/**
2 * @author huangzhang
3 * @description
4 * @date Created in 2018/7/16 19:31
5 */
6@Component
7public class Consumer {
8    @JmsListener(destination = "mytest.queue")
9    public void receiveQueue(String text){
10        System.out.println("Consumer收到的报文为:"+text);
11    }
12}

创建第二个消费者（这里不光消费了生产者插入队列中的message，而且将返回值插入到了"return.queue"队列中）：

1/**
2 * @author huangzhang
3 * @description
4 * @date Created in 2018/7/16 19:33
5 */
6@Component
7public class Consumer1 {
8    @JmsListener(destination = "mytest.queue")
9    @SendTo("return.queue")
10    public String receiveQueue(String message){
11        System.out.println("Consumer1收到的报文为:"+message);
12        return "========return message "+message;
13    }
14}

测试方法：

1@Service
2public class SpringbootJmsApplicationTests {
3    @Autowired
4    private Provider provider;
5    
6    public void contextLoads() throws InterruptedException {
7        Destination destination = new ActiveMQQueue("mytest.queue");
8        for(int i=0; i<10; i++){
9            provider.sendMessage(destination, "huangzhang "+i);
10        }
11    }
12
13}

这里我在controller中调用了测试方法：

1/**
2 * @author huangzhang
3 * @description
4 * @date Created in 2018/7/16 20:23
5 */
6@Controller
7public class Test {
8    @Autowired
9    SpringbootJmsApplicationTests springbootJmsApplicationTests;
10    @RequestMapping("/")
11    @ResponseBody
12    public String test01()throws Exception{
13        
14        springbootJmsApplicationTests.contextLoads();
15        
16        return "success!";
17    }
18}

访问http://localhost:8080/对此demo进行测试

这里是执行结果，可以看出，两个消费者分别消费了生产者放入消息队列中的消息，并且Consumer1消费者将返回结果放入了队列中供生产者消费。

查看Queues

这里可以看出我们生产者循环往mytest.queue队列中写入10次，由两个消费者消费了这10次消息

consumer1消费了5次消息，并往返回队列return.queue中写入5次，由原生产者消费了者5次消息

到这里一个简单的spring boot整合activeMQ的一对一（queue）模式的demo就完成了（请多指正）。

下面我们对provider和consumer进行改造，实现多对多（topic）模式：

对test类进行修改，修改为以下代码：

1@Service
2public class SpringbootJmsApplicationTests {
3    @Autowired
4    private Provider provider;
5    @Autowired
6    private TopicSender topicSender;
7
8    /*public void contextLoads() throws Exception {
9        Destination destination = new ActiveMQQueue("mytest.queue");
10        for(int i=0; i<10; i++){
11            provider.sendMessage(destination, "huangzhang "+i);
12        }
13    }*/
14    public void topicSend(){
15        Destination destination  = new ActiveMQTopic("test.topic");
16        for (int i = 0; i < 10 ; i++){
17            provider.sendMessage(destination, "topic"+i);
18        }
19    }
20
21}

订阅者为：

1/**
2 * @author huangzhang
3 * @description
4 * @date Created in 2018/7/16 19:31
5 */
6@Component
7public class Consumer {
8    /*@JmsListener(destination = "mytest.queue")
9    public void receiveQueue(String text){
10        System.out.println("Consumer收到的报文为:"+text);
11    }*/
12
13    @JmsListener(destination = "test.topic")
14    public void receiveQueue1(String text){
15        System.out.println("Consumer收到的----topic----报文为:"+text);
16    }
17}

两个订阅者修改方式一样。

然后像queue一样，通过controller调用test方法：

1/**
2 * @author huangzhang
3 * @description
4 * @date Created in 2018/7/10 20:23
5 */
6@Controller
7public class Test {
8    @Autowired
9    SpringbootJmsApplicationTests springbootJmsApplicationTests;
10    @RequestMapping("/")
11    @ResponseBody
12    public String test01()throws Exception{
13//        springbootJmsApplicationTests.contextLoads();
14        springbootJmsApplicationTests.topicSend();
15       
16        return "success!";
17    }
18}

同样的启动项目之后调用http://localhost:8080/接口，控制台打印如下：

我们发现，这里consumer和consumer1两个订阅者都收到了10条订阅的"test.topic"消息，再次证明：

生产者发送一条消息到queue，只有一个消费者能收到；

发布者发送到topic的消息，只要订阅了topic的订阅者就会收到消息。