RabbitMq使用方法

1.RabbitMq运行机制

发送者(Producer) --> 交换机(Exchange) --> 队列(Queues) --> 消费者(Consumer) ​​​​​​​

2.交换机(Exchange) 类型

direct（直接）：消息精确发送，而且只能到达一个队列

fanout（扇出）：消息会分发到所有绑定的队列上。

topic（主题）：与fanout类似，但是多了个属性路由key绑定，消息会发送到路由key对应的队列上

3.spring整合rabbitmq

3.1 导入amqp依赖

    org.springframework.boot
    spring-boot-starter-amqp

 3.2 yml配置rabbitmq

rabbitmq:
    host: 122.77.14.12
    port: 5672
    username: root
    password: root
    virtual-host: /

3.3 在启动类上加上@EnableRabbit注解

3.4 配置 把RabbitMQ接收和发送的数据转为json

1. AmqpAdmin：管理组件
2. RabbitTemplate：消息发送处理组件
3. @RabbitListener 监听消息的方法可以有三种参数（不分数量，顺序）
• Object content, Message message, Channel

package com.wanxin.gulimall.order.config;


import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


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

3.5 测试发送消息

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class GulimallOrderApplicationTests {

    // 可以创建交换机，队列，并管理
    @Autowired
    private AmqpAdmin amqpAdmin;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    
    @Test
    public void createExchange() {

        
        Exchange directExchange = new DirectExchange("hello-java-exchange",true,false);
        amqpAdmin.declareExchange(directExchange);
        log.info("Exchange[{}]创建成功：","hello-java-exchange");
    }


    
    @Test
    public void testCreateQueue() {
        
        Queue queue = new Queue("hello-java-queue",true,false,false);
        amqpAdmin.declareQueue(queue);
        log.info("Queue[{}]创建成功：","hello-java-queue");
    }


    
    @Test
    public void createBinding() {

        
        Binding binding = new Binding("hello-java-queue",
                Binding.DestinationType.QUEUE,
                "hello-java-exchange",
                "hello.java",
                null);
        amqpAdmin.declareBinding(binding);
        log.info("Binding[{}]创建成功：","hello-java-binding");

    }

    
    @Test
    public void sendMessageTest() {
        OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
        reasonEntity.setId(1L);
        reasonEntity.setCreateTime(new Date());
        reasonEntity.setName("reason");
        reasonEntity.setStatus(1);
        reasonEntity.setSort(2);
        String msg = "Hello World";
        //1、发送消息,如果发送的消息是个对象，会使用序列化机制，将对象写出去，对象必须实现Serializable接口

        //2、发送的对象类型的消息，可以是一个json
        
        rabbitTemplate.convertAndSend("hello-java-exchange","hello2.java",
                reasonEntity,new CorrelationData(UUID.randomUUID().toString()));
        rabbitTemplate.convertAndSend("hello-java-exchange","hello.java",
                reasonEntity,new CorrelationData(UUID.randomUUID().toString()));
        log.info("消息发送完成:{}",reasonEntity);
    }

//    @Test
//    public void create() {
//        HashMap arguments = new HashMap<>();
//        arguments.put("x-dead-letter-exchange", "order-event-exchange");
//        arguments.put("x-dead-letter-routing-key", "order.release.order");
//        arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
//        Queue queue = new Queue("order.delay.queue", true, false, false, arguments);
//        amqpAdmin.declareQueue(queue);
//        log.info("Queue[{}]创建成功：","order.delay.queue");
//    }

}

3.5 测试接收消息

1.@RabbitListener 类+方法上(监听哪些队列)

2.@RabbitHandler 标在方法上(重载区分不同的消息)

    
    @RabbitListener(queues = {"hello-java-queue"})
    public void revieveMessage(Message message, OrderReturnReasonEntity content, Channel channel) {
        //拿到主体内容
        byte[] body = message.getBody();
        //拿到的消息头属性信息
        MessageProperties messageProperties = message.getMessageProperties();
        System.out.println("接受到的消息...内容" + message + "===内容：" + content.getClass());

    }