Rabbit MQ的几种模式

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。

官网文档：https://www.rabbitmq.com/getstarted.html

Rabbit MQ有几种工作方式：

简单模式：一个生产者，一个消费者

work模式：一个生产者，多个消费者，每个消费者获取到的消息唯一，平均消费。

订阅模式：一个生产者发送的消息会被多个消费者获取。

路由模式：发送消息到交换机并且要指定路由key ，消费者将队列绑定到交换机时需要指定路由key

topic模式：将路由键和某模式进行匹配，此时队列需要绑定在一个模式上，“#”匹配一个词或多个词，“*”只匹配一个词。

rpc模式：客户端向一个队列中发送消息，并注册一个回调的队列用于接收服务端返回的消息，该消息需要声明一个叫做correaltionId的属性，

该属性将是该次请求的唯一标识。服务端在接受到消息（在需要时可以验证correaltionId）后，处理消息，并将消息发送到客户端注册的回调队列中。

1、简单模式

配置：

public final static String SIMPLE_QUEUE = "simpleQueue";

@Bean

public Queue simpleQueue() {

    return new Queue(SIMPLE_QUEUE, true);

}

生产者：

rabbitTemplate.convertAndSend(RabbitConfig.SIMPLE_QUEUE, msg);

消费者：


@RabbitListener(queues = RabbitConfig.SIMPLE_QUEUE)

public void simpleListen(String msg) {

    System.out.println("simple队列 接收到消息：" + msg);

}

2、work模式

一个生产者，多个消费者，每个消费者获取到的消息唯一。一条消息只能被其中一个消费掉，相互争夺资源。

配置：

public final static String WORK_QUEUE = "workQueue";

@Bean

public Queue workQueue() {

    return new Queue(WORK_QUEUE, true);

}

生产者：

public void sendWorkQueueMq(String msg) {

    rabbitTemplate.convertAndSend(RabbitConfig.WORK_QUEUE, msg);

    logger.info("发送消息:{}", msg);

}

消费者：

@RabbitListener(queues = RabbitConfig.WORK_QUEUE)

    public void workListen1(String msg) {

    System.out.println("work模式1 接收到消息：" + msg);

}

@RabbitListener(queues = RabbitConfig.WORK_QUEUE)

    public void workListen2(String msg) {

    System.out.println("work模式2 接收到消息：" + msg);

}

3、发布/订阅模式

一个生产者发送的消息会被多个消费者获取

配置：

public final static String FANOUT_QUEUE_ONE = "fanout_queue_one";

public final static String FANOUT_QUEUE_TWO = "fanout_queue_two";

public final static String FANOUT_EXCHANGE = "fanout_exchange";

// fanout 广播者模式队列

@Bean

public Queue fanoutQueueOne() {

    return new Queue(FANOUT_QUEUE_ONE, true);

}

@Bean

public Queue fanoutQueueTwo() {

    return new Queue(FANOUT_QUEUE_TWO, true);

}

// fanout 交换器

@Bean

public FanoutExchange fanoutExchange() {

    return new FanoutExchange(FANOUT_EXCHANGE);

}

// 广播模式绑定

@Bean

public Binding fanoutExchangeBingingOne() {

    return BindingBuilder.bind(fanoutQueueOne()).to(fanoutExchange());

}

@Bean

public Binding fanoutExchangeBingingTwo() {

    return BindingBuilder.bind(fanoutQueueTwo()).to(fanoutExchange());

}

生产者：

public void sendFanoutExchangeMq(String msg) {

    rabbitTemplate.convertAndSend(RabbitConfig.FANOUT_EXCHANGE, "", msg);

    logger.info("发送消息:{}", msg);

}

消费者：

@RabbitListener(queues = RabbitConfig.FANOUT_QUEUE_ONE)

public void fanoutListen1(String msg) {

    System.out.println("fanout模式1 接收到消息：" + msg);

}

@RabbitListener(queues = RabbitConfig.FANOUT_QUEUE_TWO)

public void fanoutListen2(String msg) {

    System.out.println("fanout模式2 接收到消息：" + msg);

}

4、路由模式

发送消息到交换机并且要指定路由key ，消费者将队列绑定到交换机时需要指定路由key。那么消息只会发送到相应key相同的队列，接着监听该队列的消费者消费消息。

配置：

public final static String DIRECT_QUEUE_ONE = "direct_queue_one";

public final static String DIRECT_QUEUE_TWO = "direct_queue_two";

public final static String DIRECT_EXCHANGE = "direct_exchange";

// direct 路由模式队列

@Bean

public Queue directQueueOne() {

    return new Queue(DIRECT_QUEUE_ONE, true);

}

@Bean

public Queue directQueueTwo() {

    return new Queue(DIRECT_QUEUE_TWO, true);

}

// direct 交换器

@Bean

public DirectExchange directExchange() {

    return new DirectExchange(DIRECT_EXCHANGE);

}

//路由模式绑定

@Bean

public Binding directExchangeBingingOne() {

    return BindingBuilder.bind(directQueueOne()).to(directExchange()).with("orange");

}

@Bean

public Binding directExchangeBingingTwo() {

    return BindingBuilder.bind(directQueueTwo()).to(directExchange()).with("black");

}

生产者：

public void sendDirectExchangeMq(String routingKey, String msg) {

    rabbitTemplate.convertAndSend(RabbitConfig.DIRECT_EXCHANGE,"orange" , msg);

    logger.info("发送消息:{}", msg);

}

消费者：

@RabbitListener(queues = RabbitConfig.DIRECT_QUEUE_ONE)

public void directListenOne(String msg) {

    System.out.println("direct模式1 接收到消息：" + msg);

}

@RabbitListener(queues = RabbitConfig.DIRECT_QUEUE_TWO)

public void directListenTwo(String msg) {

    System.out.println("direct模式2 接收到消息：" + msg);

}

如上代码，只有routingKey 为orange的能收到消息

5、topic模式

将路由键和某模式进行匹配，此时队列需要绑定在一个模式上，“#”匹配一个词或多个词，“*”只匹配一个词。

配置：

public final static String TOPIC_QUEUE_ONE = "topic_queue_one";

public final static String TOPIC_QUEUE_TWO = "topic_queue_two";

public final static String TOPIC_EXCHANGE = "topic_exchange";

public final static String TOPIC_ROUTINGKEY_ONE = "common.key";

public final static String TOPIC_ROUTINGKEY_TWO = "*.key";

// topic 订阅者模式队列

@Bean

public Queue topicQueueOne() {

    return new Queue(TOPIC_QUEUE_ONE, true);

}

@Bean

public Queue topicQueueTwo() {

    return new Queue(TOPIC_QUEUE_TWO, true);

}

// topic 交换器

@Bean

public TopicExchange topExchange() {

    return new TopicExchange(TOPIC_EXCHANGE);

}

// 订阅者模式绑定

@Bean

public Binding topicExchangeBingingOne() {

    return BindingBuilder.bind(topicQueueOne()).to(topExchange()).with(TOPIC_ROUTINGKEY_ONE);

}

@Bean

public Binding topicExchangeBingingTwo() {

    return BindingBuilder.bind(topicQueueTwo()).to(topExchange()).with(TOPIC_ROUTINGKEY_TWO);

}

生产者：

public void sendTopicExchangeMq(String routingKey, String msg) {

    rabbitTemplate.convertAndSend(RabbitConfig.TOPIC_EXCHANGE, "common.key", msg);

 logger.info("发送消息:{}", msg);

}

消费者：

@RabbitListener(queues = RabbitConfig.TOPIC_QUEUE_ONE)

public void topicListenOne(String msg) {

    System.out.println("topic模式1 接收到消息：" + msg);

}

@RabbitListener(queues = RabbitConfig.TOPIC_QUEUE_TWO)

public void topicListenTwo(String msg) {

    System.out.println("topic模式2 接收到消息：" + msg);

}

根据routingKey匹配对应的才能收到消息

6、rpc模式

客户端向一个队列中发送消息，并注册一个回调的队列用于接收服务端返回的消息，该消息需要声明一个叫做correaltionId的属性，