rabbitmq几种发送消息模式-快速入门

概念

rabbitmq概念比较多，这里说几个重要的

routing key: 消息体的一部分 发送消息时指定，用于队列和交换机绑定关系进行匹配。

exchange 交换机:rabbitmq发送消息直接发送到交换机，由交换机发送消息给队列。

queue：用于存储消息，交换机发送的消息会进入这里面，在创建的时候可以指定消息持久化，设置durable属性为TRUE。

bingKey 用于绑定交换机和队列的关系，在项目初始化的时候声明，bindkey决定了通配符的消息进入哪些队列

通配符：支持 *和# 比如 a.b.c 就是三个单词，单词多少以点为界限确定 #可以匹配多个单词*只可以匹配一个单词

简单结构图如下：

发送方式

rabbitmq支持了点对点，通配符，广播模式等发送方式。

下面主要就这三点做一个简单概述,示例代码使用springboot

点对点

发送消息指定消息routing key 为 a.b ,只有接收端交换机和队列使用a.b绑定的时候才可以使用

配置类

package com.tom.config;import com.tom.message.Demo01Message;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {/*** Direct Exchange 示例的配置类*/public static class DirectExchangeDemoConfiguration {// 创建 Queue//点对点发布@Beanpublic Queue demo01Queue() {return new Queue(Demo01Message.QUEUE, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Direct Exchange@Beanpublic DirectExchange demo01Exchange() {return new DirectExchange(Demo01Message.EXCHANGE,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding// Exchange：Demo01Message.EXCHANGE// Routing key：Demo01Message.ROUTING_KEY// Queue：Demo01Message.QUEUE@Beanpublic Binding demo01Binding() {return BindingBuilder.bind(demo01Queue()).to(demo01Exchange()).with(Demo01Message.ROUTING_KEY);}}}

消费者：

package com.tom.mq;// Demo01Consumer.javaimport com.tom.message.Demo01Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@RabbitListener(queues = Demo01Message.QUEUE,concurrency = "5")
public class Demo01Consumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(Demo01Message message) {logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);}}

发送端：

 rabbitTemplate.convertAndSend(Demo01Message.EXCHANGE, Demo01Message.ROUTING_KEY, message);

topic方式发送

配置类，在点对点的配置类的基础上加上内部类

/*** 下面的配置为通配符发布 topic发布*/public static  class LogExchangeConfiguration{// 创建 Queue@Beanpublic Queue logQueue1() {return new Queue(Constant.myQueue1, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}@Beanpublic Queue logQueue2() {return new Queue(Constant.myQueue2, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Topic Exchange@Beanpublic TopicExchange logExchange() {return new TopicExchange(Constant.myExchange,true,  // durable: 是否持久化false);  // exclusive: 是否排它}// 创建 Binding//一个queue 多个@Beanpublic Binding BindingLogQueue1() {return BindingBuilder.bind(logQueue1()).to(logExchange()).with("a.b");}@Beanpublic Binding bindingLogQueue2() {return BindingBuilder.bind(logQueue2()).to(logExchange()).with("a.*");}}

消费者：

package com.tom.mq;// Demo01Consumer.javaimport com.tom.config.Constant;
import com.tom.message.Demo01Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@RabbitListener(queues = Constant.myQueue1,concurrency = "5")
public class Consumer1 {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage( String message) {logger.info("myQueue1收到消息【{}】", message);}}

package com.tom.mq;// Demo01Consumer.javaimport com.tom.config.Constant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@RabbitListener(queues = Constant.myQueue2,concurrency = "5")
public class Consumer2 {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage( String message) {logger.info("myQueue2收到消息{}", message);}}

发送端：

package com.tom;import com.tom.config.Constant;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class TopicTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void test() {String  message="hello a--------------------->>>>>>>>>>>>>>>>>>>>>";rabbitTemplate.convertAndSend(Constant.myExchange,"a.b",message);System.out.println("发送消息success！！！！");}
}

发送结果：

 发布模式和上面两种类似，这里暂时不做赘述。

开发总结

1.声明队列，交换机，建立交换机和队列的绑定关系

2.建立监听队列的监听器

3.发送消息指定消息的routingkey

4.接收消息处理业务

代码地址：

 https://github.com/weiwensi/cloud-parent.git

地址上项目的rabbitmq模块

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！