spring集成rabbitmq

maven配置

//pom.xml
<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>

rabbitMQ配置文件

//rabbitmq-config.properties
mq.host=127.0.0.1
mq.username=test
mq.password=123456
mq.port=5672
mq.vhost=testmq
mq.queueName=test

Spring配置

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

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

    <!-- 消息对象json转换类 -->
    <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />

申明一个消息队列Queue

//application-mq.xml
<rabbit:queue id="${mq.queueName}" name="${mq.queueName}" durable="true" auto-delete="false" exclusive="false" />

• tips
  durable:是否持久化
  exclusive: 仅创建者可以使用的私有队列，断开后自动删除
  auto_delete: 当所有消费客户端连接断开后，是否自动删除队列

交换机定义

<rabbit:direct-exchange name="amqpExchange" durable="true" auto-delete="false" id="test-mq-amqpExchange">
    <rabbit:bindings>
        <rabbit:binding queue="${mq.queueName}" key="${mq.queueName}"/>
    </rabbit:bindings>
</rabbit:direct-exchange>

• tips
  rabbit:direct-exchange：定义exchange模式为direct，意思就是消息与一个特定的路由键完全匹配，才会转发。
  rabbit:binding：设置消息queue匹配的key

生成者 RabbitMQService.interface

// 定义rabbitMQ接口
public interface RabbitMQService {
    /**
     * 发送消息到指定队列
     * @param queueKey
     * @param object
     */
    public void sendDataToQueue(String queueKey, Object object);
}

• 实现接口

import com.haomo.delivery.service.RabbitMQService;
import org.jboss.logging.Logger;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * Created by yide on 2018/9/7.
 */
@Service
public class RabbitMQImpl implements RabbitMQService {
    @Autowired
    private AmqpTemplate amqpTemplate;

    private final static Logger LOGGER = Logger.getLogger(RabbitMQImpl.class);
    /* (non-Javadoc)
     * @see com.stnts.tita.rm.api.mq.MQProducer#sendDataToQueue(java.lang.String, java.lang.Object)
     */
    @Override
    public void sendDataToQueue(String queueKey, Object object) {
        try {
            amqpTemplate.convertAndSend(queueKey, object);
        } catch (Exception e) {
            LOGGER.error(e);
        }

    }
}

• tips:
  convertAndSend：将Java对象转换为消息发送到匹配Key的交换机中Exchange，由于配置了JSON转换，这里是将Java对象转换成JSON字符串的形式。原文：Convert a Java object to an Amqp Message and send it to a default exchange with a specific routing key.

controller接口调用

@Controller
public class OrderInterface {

    // 注入rabbitmq服务
    @Autowired
    RabbitMQService rabbitMQService;

    @RequestMapping(value = "/test/new",
            method = RequestMethod.POST,
            produces = "application/json;charset=UTF-8")
    @ResponseBody
    public Object createOvertime(
            HttpServletRequest request,
            @RequestHeader(value = "X-Auth-Token",required = false) String token,
            @RequestParam(value = "content") String content
    ) {
        Map<String,Object> map = new HashMap();
        map.put("content",content);
        // 存储到rabbitmq中
        rabbitMQService.sendDataToQueue("test",map);
        return "success";
    }
}

消费者

@Component
public class QueueListenter implements MessageListener {
    @Override
    public void onMessage(Message msg) {
        try{
            String content = new String(msg.getBody(), "UTF-8");
            System.out.println(content);
        }catch(Exception e){
            e.printStackTrace();
        }
    }
}

监听配置

<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">
    <rabbit:listener queues="${mq.queueName}" ref="queueListenter"></rabbit:listener>
</rabbit:listener-container>

• tips
  queues：监听的队列，多个的话用逗号（,）分隔
  ref：监听器

消费者手动(安全)删除消息

@Component
public class QueueListenter implements ChannelAwareMessageListener {
         @Override
    public synchronized void onMessage(Message message, Channel channel) throws Exception {
        try{
            String content = new String(Message.getBody(), "UTF-8");
            System.out.println(content);
          // 删除rabbit队列消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
        }catch(Exception e){
            e.printStackTrace();
        }
}

监听配置

<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" transaction-size="1000">
        <rabbit:listener queues="${mq.queueName}" ref="queueListenter"></rabbit:listener>
    </rabbit:listener-container>

参考链接:https://blog.csdn.net/JaCman/article/details/50261915
官网:http://www.rabbitmq.com/