RabbitMQ：Spring整合RabbitMQ

13F2925356653218AFBD71C70F752C7C.jpg

在前面的几篇博客里面已经把RabbitMQ的一些理论详细了说明了，在这一篇中将记录下Spring整合RabbitMQ，本文只是简单一个整合介绍，属于抛砖引玉，具体实现还需大家深入研究哈..

代码我会上传到我的码云上，如需下载请在文章的末尾寻找下载地址

1、POM引入

<!-- RabbitMQ -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>3.5.1</version>
</dependency>
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>1.4.5.RELEASE</version>
</dependency>

2、RabbitMQ配置信息

添加rabbitmq.properties配置文件

rabbit.hosts=127.0.0.1
rabbit.username=hrabbit
rabbit.password=123
rabbit.port=5672
rabbit.virtualHost=/hrabbit
# 统一XML配置中易变部分的命名
rabbit.queue=rabbitmq_test

3、添加FastJson转化类

spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列，由于fastjson的速度快于jackson,这里替换为fastjson的一个实现

package www.hrabbit.cn.configer;


import com.alibaba.fastjson.JSON;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.*;

import java.io.IOException;
import java.io.UnsupportedEncodingException;


/**
 * @Auther: hrabbit
 * @Date: 2018-07-02 下午6:35
 * @Description:
 */


public class FastJsonMessageConverter  extends AbstractJsonMessageConverter {

    private static Log log = LogFactory.getLog(FastJsonMessageConverter.class);

    private static ClassMapper classMapper = new DefaultClassMapper();

    public FastJsonMessageConverter() {
        super();
    }

    @Override
    protected Message createMessage(Object object, MessageProperties messageProperties) {
        byte[] bytes = null;
        try {
            String jsonString = JSON.toJSONString(object);
            bytes = jsonString.getBytes(getDefaultCharset());
        } catch (IOException e) {
            throw new MessageConversionException("Failed to convert Message content", e);
        }
        messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        messageProperties.setContentEncoding(getDefaultCharset());
        if (bytes != null) {
            messageProperties.setContentLength(bytes.length);
        }
        classMapper.fromClass(object.getClass(), messageProperties);
        return new Message(bytes, messageProperties);
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        Object content = null;
        MessageProperties properties = message.getMessageProperties();
        if (properties != null) {
            String contentType = properties.getContentType();
            if (contentType != null && contentType.contains("json")) {
                String encoding = properties.getContentEncoding();
                if (encoding == null) {
                    encoding = getDefaultCharset();
                }
                try {
                    Class<?> targetClass = getClassMapper().toClass(message.getMessageProperties());
                    content = convertBytesToObject(message.getBody(), encoding, targetClass);
                } catch (IOException e) {
                    throw new MessageConversionException("Failed to convert Message content", e);
                }
            } else {
                log.warn("Could not convert incoming message with content-type [" + contentType + "]");
            }
        }
        if (content == null) {
            content = message.getBody();
        }
        return content;
    }

    private Object convertBytesToObject(byte[] body, String encoding, Class<?> clazz)
            throws UnsupportedEncodingException {
        String contentAsString = new String(body, encoding);
        return JSON.parseObject(contentAsString, clazz);
    }
}

4、添加amqp-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:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
    http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" >

    <description>rabbitmq 连接服务配置</description>

    <!-- 连接配置 -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbit.hosts}" username="${rabbit.username}" password="${rabbit.password}" port="${rabbit.port}"  virtual-host="${rabbit.virtualHost}"/>
    <rabbit:admin connection-factory="connectionFactory"/>

    <!-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列，由于fastjson的速度快于jackson,这里替换为fastjson的一个实现 -->
    <bean id="jsonMessageConverter"  class="www.hrabbit.cn.util.FastJsonMessageConverter"></bean>

    <!-- spring template声明-->
    <rabbit:template exchange="koms" id="amqpTemplate"  connection-factory="connectionFactory" message-converter="jsonMessageConverter"/>

    <!--
        durable:是否持久化

        exclusive: 仅创建者可以使用的私有队列，断开后自动删除

        auto_delete: 当所有消费客户端连接断开后，是否自动删除队列
     -->

    <!--  申明一个消息队列Queue   -->
    <rabbit:queue id="order" name="order" durable="true" auto-delete="false" exclusive="false" />
    <!--
     rabbit:direct-exchange：定义exchange模式为direct，意思就是消息与一个特定的路由键完全匹配，才会转发。

    rabbit:binding：设置消息queue匹配的key
     -->
    <!-- 交换机定义 -->
    <rabbit:direct-exchange name="koms" durable="true" auto-delete="false" id="koms">
        <rabbit:bindings>
            <rabbit:binding queue="order" key="order"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!--
         queues：监听的队列，多个的话用逗号（,）分隔

        ref：监听器
     -->
    <bean class="www.hrabbit.cn.rabbitMq.listener.MessageListener" id="messageListener"></bean>
    <!-- 配置监听  acknowledeg = "manual"   设置手动应答  当消息处理失败时:会一直重发  直到消息处理成功 -->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
        <!-- 配置监听器 -->
        <rabbit:listener queues="order" ref="messageListener"/>
    </rabbit:listener-container>
</beans>

在这个项目中我的生产者和消费者是放到同一个项目中的。项目中的监听器，即为消费者。

5、生产者

注入AmqpTemplate模板，调用convertAndSend ();方法添加消息；

package www.hrabbit.cn.rabbitMq.service.impl;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;
import www.hrabbit.cn.rabbitMq.service.SpittleService;

import javax.annotation.Resource;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * @Auther: hrabbit
 * @Date: 2018-07-02 下午4:26
 * @Description:
 */
@Service("spittleService")
public class SpittleServiceImpl implements SpittleService {

    @Resource
    private AmqpTemplate amqpTemplate;

    /**
     * 生产消息
     * @return
     */
    public Map<String,Object> spittleMsg(){

        Map<String,Object> dataList = new LinkedHashMap<>();

        for (int i=0;i<10;i++){
            dataList.put("order","msgResult:"+i);
            amqpTemplate.convertAndSend("order","msgResult:"+i);
        }
        return dataList;
    }

}

6、添加监听器(即消费者)

package www.hrabbit.cn.rabbitMq.listener;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;

/**
 * @Auther: hrabbit
 * @Date: 2018-07-02 下午4:47
 * @Description:
 */
@Component
public class MessageListener implements ChannelAwareMessageListener {

    private Logger logger= LoggerFactory.getLogger(MessageListener.class);

    @Transactional
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        //业务处理，放到action层，并返回处理成功还是异常的flag
        //boolean mqFlag=rabbitMaConsumerTaskAction.saveMq(arg0);
        //还有一个点就是如何获取mq消息的报文部分message？
        String result=new String(message.getBody(),"UTF-8");
        System.out.println("消息:"+result);
        if(true){
            basicACK(message,channel);//处理正常--ack
        }else{
            basicNACK(message,channel);//处理异常--nack
        }
    }


    //正常消费掉后通知mq服务器移除此条mq
    private void basicACK(Message message,Channel channel){
        try{
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }catch(IOException e){
            logger.error("通知服务器移除mq时异常，异常信息："+e);
        }
    }
    //处理异常，mq重回队列
    private void basicNACK(Message message,Channel channel) {
        try {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        } catch (IOException e) {
            logger.error("mq重新进入服务器时出现异常，异常信息：" + e);
        }
    }
}

7、启动项目，测试

访问地址:http://localhost:8080/amqp/spittleMsg生产了10条消息，此时查看控制台10条消息都被消费了！

image.png

项目地址:https://gitee.com/hrabbit/spring-rabbitMQ

系列文章：

RabbitMQ：RabbitMQ-理论基础
RabbitMQ：RabbitMQ：快速入门hello word
RabbitMQ：RabbitMQ：work queues 工作队列(Round-robin/Fair dispatch)
RabbitMQ：RabbitMQ：消息应答与消息持久化
RabbitMQ：发布/订阅 Publish/Subscribe
RabbitMQ：路由Routing
RabbitMQ：Topic类型的exchange
RabbitMQ：RabbitMQ之消息确认机制（事务+Confirm）