activeMQ入门+spring boot整合activeMQ

时间：2018-07-18 10:51:08 阅读：158 评论：0 收藏：0 [点我收藏+]

标签：rabbit connect 消费会话 print rocketmq factory 插入 inter

最近想要学习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中引用

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <!-- <version>5.7.0</version> -->
        </dependency>

在application.properties中配置activeMQ连接：

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

创建消息生产者：

/**
 * @author huangzhang
 * @description
 * @date Created in 2018/7/16 18:57
 */
@Service("provider")
public class Provider {
    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;

    public void sendMessage(Destination destination, final String message){
        jmsMessagingTemplate.convertAndSend(destination,message);

    }
　　//消费消费者返回的队列"return.queue"中的消息
    @JmsListener(destination="return.queue")
    public void consumerMessage(String string){
        System.out.println("从return.queue队列收到的回复报文为:"+string);
    }

}

创建第一个消费者：

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

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

/**
 * @author huangzhang
 * @description
 * @date Created in 2018/7/16 19:33
 */
@Component
public class Consumer1 {
    @JmsListener(destination = "mytest.queue")
    @SendTo("return.queue")
    public String receiveQueue(String message){
        System.out.println("Consumer1收到的报文为:"+message);
        return "========return message "+message;
    }
}

测试方法：

@Service
public class SpringbootJmsApplicationTests {
    @Autowired
    private Provider provider;
    
    public void contextLoads() throws InterruptedException {
        Destination destination = new ActiveMQQueue("mytest.queue");
        for(int i=0; i<10; i++){
            provider.sendMessage(destination, "huangzhang "+i);
        }
    }

}

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

/**
 * @author huangzhang
 * @description
 * @date Created in 2018/7/16 20:23
 */
@Controller
public class Test {
    @Autowired
    SpringbootJmsApplicationTests springbootJmsApplicationTests;
    @RequestMapping("/")
    @ResponseBody
    public String test01()throws Exception{
        
        springbootJmsApplicationTests.contextLoads();
        
        return "success!";
    }
}