RabbitMQ 消息队列使用

MQ

同步调用和异步调用

同步调用优点:

时效性强,立即得到结果

缺点:

• 耦合度高 新业务新需求到来时,需要修改代码
• 性能和吞吐能力下降 调用服务的响应时间为所有服务的时间之和
• 资源浪费 调用链中的服务在等待时不会释放请求占用的资源
• 级联失败 一个服务执行失败会导致调用链后续所有服务失败

异步调用优点:

• 服务解耦 便于扩展
• 性能提高 吞吐量提高
• 不会级联失败
• 流量削峰

RabbitMQ

基础概念

• channel: 操作MQ工具
• exchange: 交换机, 将消息路由到队列中
• queue: 保存消息的队列
• virtual host: 虚拟主机, 相当于namespace,隔离的环境,对queue和exchange的逻辑分组

模型

基于 Spring Amqp

引入依赖

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

简单队列 (Hello-World)

消息发送者和接收者都需要以下配置:

spring:
  rabbitmq:
    port: 5672
    host: localhost
    virtual-host: /
    username: guest
    password: guest

发送消息:

@SpringBootTest
@RunWith(SpringRunner.class)
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void helloWorldModel(){
        String queueName="zbq.queue1";
        String message="hello, spring amqp";
        rabbitTemplate.convertAndSend(queueName,message);
    }
}

消息接受:

@Component
@Slf4j
public class SpringRabbitListener {
    @RabbitListener(queues = "zbq.queue1")
    public void listenSimpleQueue(String msg){
        log.info("收到消息: "+msg);
    }
}

工作队列 (Work Queue)

发送者:

@Test
    public void workQueueModel(){
        String queueName="zbq.work.queue";
        String msg="hello, amqp ";
        for(int i=0;i<1000;i++){
            rabbitTemplate.convertAndSend(queueName,msg+i);
        }
    }

接收者:

@RabbitListener(queues = "zbq.work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    log.info("消费者1号接收到消息: "+msg);
    Thread.sleep(20);
}

@RabbitListener(queues = "zbq.work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    log.info("消费者2号接收到消息: "+msg);
    Thread.sleep(100);
}

Pub/Sub (Fanout exchange)

Fanout交换将将消息发送到每一个绑定到它的队列中

1. 声明一个FanoutExchange,声明2个队列, 绑定队列到FanoutExchange上

@Configuration
public class FanoutConfig {
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("zbq.fanout");
    }

    @Bean
    public Queue fanoutQueue1(){
        return new Queue("zbq.fanout.queue1");
    }

    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    @Bean
    public Queue fanoutQueue2(){
        return new Queue("zbq.fanout.queue2");
    }

    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

2.消费者监听这两个队列

@RabbitListener(queues = "zbq.fanout.queue1")
public void listenFanoutQueue1(String msg){
    log.info("消费者1收到Fanout消息: "+msg);
}

@RabbitListener(queues = "zbq.fanout.queue2")
public void listenFanoutQueue2(String msg){
    log.info("消费者2收到Fanout消息: "+msg);
}

3.发消息到fanoutexchange

@Test
public void fanoutModel(){
    String exchangeName="zbq.fanout";
    String msg="hello, fanout ";
    for(int i=0;i<10;i++){
        rabbitTemplate.convertAndSend(exchangeName,"",msg+i);
    }
}

查看消费者输出信息

Direct Exchange

Direct交换机会将消息按照路由规则发送到指定的队列

1.声明交换机, 队列,并绑定,添加routingkey

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.direct.queue1"),
    exchange = @Exchange(name="zbq.direct",type = ExchangeTypes.DIRECT),
    key = {"girlfriend","family"}
))
public void listenDirectQueue1(String msg){
    log.info("消费者1收到Direct消息: "+msg);
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.direct.queue2"),
    exchange = @Exchange(name="zbq.direct",type = ExchangeTypes.DIRECT),
    key = {"friend","family"}
))
public void listenDirectQueue2(String msg){
    log.info("消费者2收到Direct消息: "+msg);
}

2.发送消息

@Test
public void directModel(){
    String exchangeName="zbq.direct";
    String msg="晚上回去吃饭 ";
    for (int i = 0; i < 10; i++) {
        rabbitTemplate.convertAndSend(exchangeName,"family",msg+i);
    }
}

@Test
public void directModel2(){
    String exchangeName="zbq.direct";
    String msg="hello, direct";
    for (int i = 0; i < 10; i++) {
        rabbitTemplate.convertAndSend(exchangeName,"girlfriend",msg+i);
    }
}

Topic Exchange

话题交换机的routingkey 必须是多个单词的列表,并以.分隔

可以使用通配符#和*

#:代表0个或者多个单词

*:代表1个单词

1.定义

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.topic.queue1"),
    exchange = @Exchange(name="zbq.topic",type = ExchangeTypes.TOPIC),
    key = "China.#"
))
public void listenTopicQueue1(String msg){
    log.info("消费者1收到Topic消息: "+msg);
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "zbq.topic.queue2"),
    exchange = @Exchange(name="zbq.topic",type = ExchangeTypes.TOPIC),
    key = "#.weather"
))
public void listenTopicQueue2(String msg){
    log.info("消费者2收到Topic消息: "+msg);
}

2.发消息

@Test
public void topicModel(){
    String exchangeName="zbq.topic";
    String msg="首都北京,今日气温10摄氏度";
    rabbitTemplate.convertAndSend(exchangeName,"China.weather",msg);
}

序列化方式

@Test
public void test(){
    Map<String,Object> map=new HashMap<>();
    map.put("hair","long");
    map.put("eyes","big");
    rabbitTemplate.convertAndSend("zbq.queue1",map);
}

发送对象类型过去, 查看序列化后的值

RabbitMQ默认使用JDK自带序列化

引入以下依赖修改序列化方法:

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

注入Bean

@Bean
public MessageConverter customMC(){
    return new Jackson2JsonMessageConverter();
}