Spring Boot----SpringBoot 整合 RabbiMQ

首先了解我的这篇博客：https://www.cnblogs.com/yanxiaoge/p/11379715.html（下面的基于这篇博客中的配置）

1、创建项目

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring‐boot‐starter‐amqp</artifactId>
</dependency>

自动配置

1、RabbitAutoConfiguration

2、有自动配置了连接工厂ConnectionFactory；

3、RabbitProperties 封装了RabbitMQ的配置

4、RabbitTemplate：给RabbitMQ发送和接受消息；

5、AmqpAdmin:RabbitMQ系统管理功能组件

2、使用

2.1 application.properties

#host、port、virtual-host默认值都是下面配置的（可以不写）
spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=5672  //服务间通讯端口默认是5672
spring.rabbitmq.virtual-host=/

2.2 测试消息的发送和接受

    @Autowired
    public RabbitTemplate rabbitTemplate;

    @Test
    public void contextLoads() {
        //Message需要自己构造一个；定义消息体内容和消息头
        //rabbitTemplate.send(exchange,routingKey,message);

        //object默认当做消息体,只需要传入要发送的object,自动序列化发送给rabbitmq
        //rabbitTemplate.convertAndSend(exchange,routingKey,Object);

        HashMap<String, String> hashMap = new HashMap<>();
        hashMap.put("1","1");
        hashMap.put("2","2");
        rabbitTemplate.convertAndSend("exchange.direct","queue1",hashMap);  //指定的exchange.direct是配置的点对点的，如果其他的可以自己测试
    }
    @Test
    public void test2() {
        HashMap<String,String> queue1 = (HashMap<String, String>) rabbitTemplate.receiveAndConvert("queue1"); //注意如果取出来的数据不是HashMap，不能强转
        System.out.println(queue1.get("1"));
    }

2.3 自定义序列化转换器，将数据以json数据存入到队列中，测试方法按照上面的测试（如果报错某个class找不到，可以自己导入 jackson-databind这个jar包）

@Configuration
public class MyAMQPConfig {
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

2.4 监听消息队列

2.4.1 @EnableRabbit  

@EnableRabbit  //开启注解的 RabbitMQ 模式
@SpringBootApplication
public class Springboot11Application {
    public static void main(String[] args) {
        SpringApplication.run(Springboot11Application.class, args);
    }
}

2.4.2 @RabbitListener

package com.zy.springboot11.service;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;


import java.util.HashMap;

@Service
public class HashMapService {
    @RabbitListener(queues = "queue1")
    public void receive(HashMap<String,String> hashMap){
        System.out.println("收到消息"+hashMap.toString());
    }

    @RabbitListener(queues = "queue2")
    public void receive2(Message message){
        System.out.println(message.getMessageProperties());
        System.out.println(message.getBody());
    }
}

2.4.3 测试

方式1、开启服务器，通过http://localhost:15672/#/queues 给消息队列发送消息

方式2:、直接通过springboot的测试方法，不用开启服务器，（测试方法中的监听器就会将数据读取出来）

2.5 代码创建Exchange、queue、绑定规则

    @Autowired
    public AmqpAdmin amqpAdmin;

    @Test
    public void create(){
        //创建Exchange
        //Exchage可以设置durable、autoDelete等
        //amqpAdmin.declareExchange(new DirectExchange("testExchage.direct"));

        //创建queue
        //amqpAdmin.declareQueue(new Queue("test.queue"));

        //绑定
        //amqpAdmin.declareBinding(new Binding("test.queue",Binding.DestinationType.QUEUE,"testExchage.direct","test.queue",null));

        //删除绑定
        //amqpAdmin.removeBinding(new Binding("test.queue",Binding.DestinationType.QUEUE,"testExchage.direct","test.queue",null));

        //删除queue(绑定关系就解除了)
        //amqpAdmin.deleteQueue("test.queue");

        //删除exchange
        //amqpAdmin.deleteExchange("testExchage.direct");
    }

生产者

aplication.yml

server:
  port: 44000
spring:
  application:
    name: test-rabbitmq-producer
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtualHost: /

config（为了让生产者程序启动的时候，像RabbitMQ中声明队列和交换机，以及他们之间的绑定关系，如果RabbitMQ中手动已经配置好了，这个config也可以不需要了）

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitmqConfig {
    public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    public static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";
    public static final String ROUTINGKEY_EMAIL="inform.#.email.#";
    public static final String ROUTINGKEY_SMS="inform.#.sms.#";

    //声明交换机
    @Bean(EXCHANGE_TOPICS_INFORM)
    public Exchange EXCHANGE_TOPICS_INFORM(){
        //durable(true) 持久化，mq重启之后交换机还在
        return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
    }

    //声明QUEUE_INFORM_EMAIL队列
    @Bean(QUEUE_INFORM_EMAIL)
    public Queue QUEUE_INFORM_EMAIL(){
        return new Queue(QUEUE_INFORM_EMAIL);
    }
    //声明QUEUE_INFORM_SMS队列
    @Bean(QUEUE_INFORM_SMS)
    public Queue QUEUE_INFORM_SMS(){
        return new Queue(QUEUE_INFORM_SMS);
    }

    //ROUTINGKEY_EMAIL队列绑定交换机，指定routingKey
    @Bean
    public Binding BINDING_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue,
                                              @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs();
    }
    //ROUTINGKEY_SMS队列绑定交换机，指定routingKey
    @Bean
    public Binding BINDING_ROUTINGKEY_SMS(@Qualifier(QUEUE_INFORM_SMS) Queue queue,
                                          @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs();
    }
}

测试

@SpringBootTest
@RunWith(SpringRunner.class)
public class Producer05_topics_springboot {
    @Autowired
    RabbitTemplate rabbitTemplate;

    //使用rabbitTemplate发送消息
    @Test
    public void testSendEmail(){

        String message = "send email message to user";
        /**
         * 参数：
         * 1、交换机名称
         * 2、routingKey
         * 3、消息内容
         */
        rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_TOPICS_INFORM,"inform.email",message);

    }

    //使用rabbitTemplate发送消息
    @Test
    public void testSendPostPage(){

        Map message = new HashMap<>();
        message.put("pageId","5a795ac7dd573c04508f3a56");
        //将消息对象转成json串
        String messageString = JSON.toJSONString(message);
        //路由key，就是站点ID
        String routingKey = "5a751fab6abb5044e0d19ea1";
        /**
         * 参数：
         * 1、交换机名称
         * 2、routingKey
         * 3、消息内容
         */
        rabbitTemplate.convertAndSend("ex_routing_cms_postpage",routingKey,messageString);

    }
}

消费者

　　让方法来监听队列

aplication.yml

server:
  port: 44001
spring:
  application:
    name: test-rabbitmq-producer
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtualHost: /

config（为了让消费者程序启动的时候，像RabbitMQ中声明队列和交换机，以及他们之间的绑定关系，如果RabbitMQ中手动已经配置好了，这个config也可以不需要了）

 　　略（使用生产者中config配置）

@Component
public class Consumer {
    //监听队列
    @RabbitListener(queues = {RabbitmqConfig.QUEUE_INFORM_EMAIL})
    public void sendenail(String msg, Message nessage, Channel channel) {
        System.out.println("receive nessage is:" + msg);
    }
}

　　

使用@RabbitListener(queues = "queue2")监听队列，默认是单线程监听队列

在config配置中添加SimpleRabbitListenerContainerFactory（Bean） ,设置多线程

@Configuration
public class RabbitMQConfig {

    public static final String EX_MEDIA_PROCESSTASK = "ex_media_processor";

    //视频处理队列
    @Value("${xc-service-manage-media.mq.queue-media-video-processor}")
    public  String queue_media_video_processtask;

    //视频处理路由
    @Value("${xc-service-manage-media.mq.routingkey-media-video}")
    public  String routingkey_media_video;

    //消费者并发数量
    public static final int DEFAULT_CONCURRENT = 10;

    @Bean("customContainerFactory")
    public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConcurrentConsumers(DEFAULT_CONCURRENT);
        factory.setMaxConcurrentConsumers(DEFAULT_CONCURRENT);
        configurer.configure(factory, connectionFactory);
        return factory;
    }

    /**
     * 交换机配置
     * @return the exchange
     */
    @Bean(EX_MEDIA_PROCESSTASK)
    public Exchange EX_MEDIA_VIDEOTASK() {
        return ExchangeBuilder.directExchange(EX_MEDIA_PROCESSTASK).durable(true).build();
    }
    //声明队列
    @Bean("queue_media_video_processtask")
    public Queue QUEUE_PROCESSTASK() {
        Queue queue = new Queue(queue_media_video_processtask,true,false,true);
        return queue;
    }
    /**
     * 绑定队列到交换机 .
     * @param queue    the queue
     * @param exchange the exchange
     * @return the binding
     */
    @Bean
    public Binding binding_queue_media_processtask(@Qualifier("queue_media_video_processtask") Queue queue, @Qualifier(EX_MEDIA_PROCESSTASK) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(routingkey_media_video).noargs();
    }
}

 @RabbitListener配置

@RabbitListener(queues = "${xc-service-manage-media.mq.queue-media-video-processor}", containerFactory = "customContainerFactory")
public void receiveMediaProcessTask(String msg) {}