SpringBoot入门建站全系列（十七）整合ActiveMq（

2019-07-16 本文已影响4人逍遥天扬

SpringBoot入门建站全系列（十七）整合ActiveMq（JMS类消息队列）

一、概述

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间的通信。对于消息中间件，常见的角色大致也就有Producer（生产者）、Consumer（消费者）

常见的消息中间件产品：

（1）ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。这里介绍的是ActiveMQ的使用。

（2）RabbitMQ

AMQP协议的领导实现，支持多种场景。淘宝的MySQL集群内部有使用它进行通讯，OpenStack开源云平台的通信组件，最先在金融行业得到运用。

（3）ZeroMQ

史上最快的消息队列系统

（4）Kafka

Apache下的一个子项目。特点：高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统。适合处理海量数据。

Jms

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

AMQP

AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。意味着我们可以使用Java的AMQP provider，同时使用一个python的producer加一个rubby的consumer。从这一点看，AQMP可以用http来进行类比，不关心实现的语言，只要大家都按照相应的数据格式去发送报文请求，不同语言的client均可以和不同语言的server链接。

首发地址：
品茗IT-同步发布

品茗IT 提供在线支持：

一键快速构建Spring项目工具

一键快速构建SpringBoot项目工具

一键快速构建SpringCloud项目工具

一站式Springboot项目生成

Mysql一键生成Mybatis注解Mapper

代码可以在SpringBoot组件化构建https://www.pomit.cn/java/spring/springboot.html中的ActiveMQ组件中查看，并下载。

二、配置

本文假设你已经引入spring-boot-starter-web。已经是个SpringBoot项目了,如果不会搭建，可以打开这篇文章看一看《SpringBoot入门建站全系列（一）项目建立》。

2.1 Maven依赖

使用activemq可以使用spring-boot-starter-activemq，方便快捷，一般springboot对大多数开源项目都做了整合，提供了专用的stater。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

2.2 配置文件

在application.properties 中需要配置activemq的信息，也可以配置自定义的配置，如：

spring.activemq.broker-url=tcp://localhost:61616
#spring.activemq.user=admin
#spring.activemq.password=secret

jms.destQueueName=destQueue

这里面，

spring.activemq.broker-url是springboot自动装配的配置，activemq的地址。
spring.activemq.user是springboot自动装配的配置，activemq的用户名，一般自己测试都不会去设置用户名密码的。
spring.activemq.password是springboot自动装配的配置，，activemq的密码，一般自己测试都不会去设置用户名密码的。
jms.destQueueName，这是我自己定义的一个目的队列名。

三、ActiveMQ的使用

3.1 配置数据转换

ActiveMQ是分为生产者和消费者的，生产者生产的消息，如何能够被消费者正常解析，需要开发者自己对数据转换做定义，如果你非要说，我用字符串也可以啊，那当这个不存在就行了。

下面这个配置是指明了ActiveMQ的数据转换是用MappingJackson2MessageConverter，将json数据转换为对象，或者将对象转换为json。

ActiveMQConfig :

package com.cff.springbootwork.activemq;

import javax.jms.ConnectionFactory;

import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.MessageType;

@Configuration
public class ActiveMQConfig {

    @Bean
    public JmsListenerContainerFactory<?> myFactory(ConnectionFactory connectionFactory,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setMessageConverter(jacksonJmsMessageConverter());
        return factory;
    }

    @Bean
    public MessageConverter jacksonJmsMessageConverter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }
}

3.2 生产者

ActiveMQ当然需要生产者来生产信息，然后才发送到消息队列的。

JmsProducer：

package com.cff.springbootwork.activemq.handler;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

import com.cff.springbootwork.activemq.model.DefaultMqModel;

@Component
public class JmsProducer {
    @Autowired
    JmsTemplate jmsTemplate;

    @Value("${jms.destQueueName}")
    String destQueueName;

    public void send(DefaultMqModel defaultMqModel) {
        jmsTemplate.convertAndSend(destQueueName, defaultMqModel);
    }
}

这里的JmsTemplate 是Spring自动生成的bean。destQueueName注入的是配置文件中定义的目的队列。然后发送数据。

测试发送数据,ActiveMQRest :

package com.cff.springbootwork.activemq.web;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import com.cff.springbootwork.activemq.handler.JmsProducer;
import com.cff.springbootwork.activemq.model.DefaultMqModel;
import com.cff.springbootwork.activemq.model.SeccondMqModel;

@RestController
@RequestMapping("/activemq")
public class ActiveMQRest {
    @Autowired
    JmsProducer jmsProducer;

    @RequestMapping(value = "/test", method = { RequestMethod.GET })
    public DefaultMqModel test() {
        DefaultMqModel defaultMqModel = new DefaultMqModel();
        defaultMqModel.setContent("hahahahahahhahaha哈啊哈哈");
        defaultMqModel.setTitle("测试");
        defaultMqModel.setType(1);
        jmsProducer.send(defaultMqModel);
        return defaultMqModel;
    }

    @RequestMapping(value = "/test2", method = { RequestMethod.GET })
    public SeccondMqModel test2() {
        SeccondMqModel seccondMqModel = new SeccondMqModel();
        seccondMqModel.setContent("asdasdasd哈啊哈哈");
        seccondMqModel.setTitle("测试2");
        seccondMqModel.setRemark("第二个");
        seccondMqModel.setType(2);
        jmsProducer.send(seccondMqModel);
        return seccondMqModel;
    }
}

3.3 消费者

消费者监听指定的队列，需要用@JmsListener注解标明它是一个消费者。

JmsConsumer :

package com.cff.springbootwork.activemq.handler;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import com.cff.springbootwork.activemq.model.DefaultMqModel;
import com.cff.springbootwork.activemq.service.BusinessSerivce;

@Component
public class JmsConsumer {

    @Autowired
    BusinessSerivce businessSerivce;

    @JmsListener(destination = "${jms.destQueueName}")
    public void processMessage(DefaultMqModel defaultMqModel) {
        businessSerivce.doBusiness(defaultMqModel);
    }
}

这里，@JmsListener注解表明它在监听我们配置文件中配置的队列，方法的参数可以直接传入对象，因为我们在3.1中配置了统一的MessageConverter，它可以自动解析成对象。

BusinessSerivce是我自定义的一个业务逻辑处理service.

BusinessSerivce:

package com.cff.springbootwork.activemq.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.cff.springbootwork.activemq.model.DefaultMqModel;
import com.cff.springbootwork.activemq.model.SeccondMqModel;

@Service
public class BusinessSerivce {
    private final Logger log = LoggerFactory.getLogger(this.getClass());

    public void doBusiness(DefaultMqModel defaultMqModel) {
        log.info(defaultMqModel.toString());
        
        if(defaultMqModel.getType() == 2 && defaultMqModel instanceof SeccondMqModel){
            SeccondMqModel seccondMqModel = (SeccondMqModel) defaultMqModel;
            log.info(seccondMqModel.remark);
        }
    }
}