当前位置: 首页 > article >正文

rabbitmq的使用介绍

一.队列工作模式介绍

1.WorkQueues模型

生产者直接把消息发送给队列,然后消费者订阅队列

特点: 消息不会重复, 分配给不同的消费者.

代码实现:

消费者代码:

@Component
@Slf4j
public class SpringRabbitListener {@RabbitListener(queues = "work.queue")public void listenWorkQueue1(String msg) throws InterruptedException {System.out.println("消费者1接收到消息:【" + msg + "】" + LocalTime.now());Thread.sleep(20);}
}

 生产者代码:

@Test
public void testWorkQueue() throws InterruptedException {// 队列名称String queueName = "work.queue";// 消息String message = "hello, message_";for (int i = 0; i < 50; i++) {// 发送消息,每20毫秒发送一次,相当于每秒发送50条消息rabbitTemplate.convertAndSend(queueName, message + i);Thread.sleep(20);}
}

2.含有交换机的模型

Publisher:生产者,不再发送消息到队列中,而是发给交换机
Exchange:交换机,一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。
Queue:消息队列也与以前一样,接收消息、缓存消息。不过队列一定要与交换机绑定。
Consumer:消费者,与以前一样,订阅队列,没有变化

交换机的四种类型:

①Fanout:广播,将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
②Direct:订阅,基于RoutingKey(路由key)发送给订阅了消息的队列
③Topic:通配符订阅,与Direct类似,只不过RoutingKey可以使用通配符
④Headers:头匹配,基于MQ的消息头匹配(用的较少,这里不讲)

RoutingKey: 路由键.⽣产者将消息发给交换器时, 指定的⼀个字符串, ⽤来告诉交换机应该如何处理这个消息.

Binding Key:绑定. RabbitMQ中通过Binding(绑定)将交换器与队列关联起来, 在绑定的时候⼀般会指定⼀个Binding Key, 这样RabbitMQ就知道如何正确地将消息路由到队列了.

①fanout交换机

交换机和队列的关系图:

1)  可以有多个队列
2)  每个队列都要绑定到Exchange(交换机)
3)  生产者发送的消息,只能发送到交换机
4)  交换机把消息发送给绑定过的所有队列
5)  订阅队列的消费者都能拿到消息

代码实现: 

消费者代码:

@Component
@Slf4j
public class SpringRabbitListener {@RabbitListener(queues = "fanout.queue1")public void listenFanoutQueue1(String msg) {System.out.println("消费者1接收到Fanout消息:【" + msg + "】");}@RabbitListener(queues = "fanout.queue2")public void listenFanoutQueue2(String msg) {System.out.println("消费者2接收到Fanout消息:【" + msg + "】");}
}

生产者代码:

@Test
public void testFanoutExchange() {// 交换机名称String exchangeName = "hmall.fanout";// 消息String message = "hello, everyone!";rabbitTemplate.convertAndSend(exchangeName, "", message);
}
②direct交换机

交换和队列的关系图:交换机和队列之间多了一个绑定关系

1)队列与交换机的绑定,不能是任意绑定了,而是要指定一个 RoutingKey(路由key)。
2)消息的发送方在 向 Exchange发送消息时,也必须指定消息的 RoutingKey。
3)Exchange不再把消息交给每一个绑定的队列,而是根据消息的 Routing Key进行判断,只有队列的 Routingkey 与消息的 Routing key 完全一致,才会接收到消息。

代码实现:

消费者代码:

@Component
@Slf4j
public class SpringRabbitListener {@RabbitListener(queues = "direct.queue1")public void listenDirectQueue1(String msg) {System.out.println("消费者1接收到direct.queue1的消息:【" + msg + "】");}@RabbitListener(queues = "direct.queue2")public void listenDirectQueue2(String msg) {System.out.println("消费者2接收到direct.queue2的消息:【" + msg + "】");}
}

生产者代码:

@Test
public void testSendDirectExchange() {// 交换机名称String exchangeName = "hmall.direct";// 消息String message = "红色警报!日本乱排核废水,导致海洋生物变异,惊现哥斯拉!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "yellow", message);
}
③topic交换机

交换机和队列的关系图:交换机和队列之间的绑定的字符串含有通配符

通配符规则:

   # :匹配一个或多个词
   * :匹配不多不少恰好1个词

假如此时publisher发送的消息使用的`RoutingKey`共有四种:

    china.news:可以路由到队列1和2
    china.weather:可以路由到队列1
    japan.news:可以路由到队列2

代码实现:

消费者代码:

@Component
@Slf4j
public class SpringRabbitListener {@RabbitListener(queues = "topic.queue1")public void listenTopicQueue1(String msg){System.out.println("消费者1接收到topic.queue1的消息:【" + msg + "】");}@RabbitListener(queues = "topic.queue2")public void listenTopicQueue2(String msg){System.out.println("消费者2接收到topic.queue2的消息:【" + msg + "】");}
}

生产者代码: 

@Test
public void testSendTopicExchange() {// 交换机名称String exchangeName = "hmall.topic";// 消息String message = "喜报!孙悟空大战哥斯拉,胜!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

二.使用java代码声明交换机,队列,绑定交换机队列

1.基于Bean声明:写成一个配置类

@Configuration
public class FanoutConfiguration {/*** 声明交换机* @return Fanout类型交换机*/@Beanpublic FanoutExchange fanoutExchange(){//return ExchangeBuilder.fanoutExchange("hmall.fanout").build();return new FanoutExchange("hmall.fanout");}/*** 第1个队列*/@Beanpublic Queue fanoutQueue1(){//return QueueBuilder.durable("fanout.queue1").build();return new Queue("fanout.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/@Beanpublic Queue fanoutQueue2(){//return QueueBuilder.durable("fanout.queue2").build();return new Queue("fanout.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

2.使用注解声明,把声明写在消费者的注解上

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消费者1接收到direct.queue1的消息:【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}
))    

三.消息转换器

上面 convertAndSend 方法的第三个参数是一个 Object 对象,在数据传输时,spring 会把你发送的消息序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为Java对象。
只不过,默认情况下Spring采用的序列化方式是JDK序列化。众所周知,JDK序列化存在下列问题:

① 数据体积过大
② 有安全漏洞
③ 可读性差

所以我们需要配置json转换器

①引入依赖
<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>
②添加配置
@Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息jackson2JsonMessageConverter.setCreateMessageIds(true);  // 这个代码是从幂等性添加进来的return jackson2JsonMessageConverter;
}

配置消息转换器,在 publisher 和 consumer 两个服务的启动类中添加一个Bean即可。

http://www.lryc.cn/news/2386161.html

相关文章:

  • 前端的core-js是什么?有什么作用?
  • 【Python 命名元祖】collections.namedtuple 学习指南
  • 系统编程day04
  • java 加密算法的简单使用
  • Arduino Uno KY-037声音传感器实验
  • 机器学习---各算法比较
  • 基于音频Transformer与动作单元的多模态情绪识别算法设计与实现(在RAVDESS数据集上的应用)
  • Flink SQL 计算实时指标同比的实现方法
  • 什么是VR实景?有哪些高价值场景?
  • 基于MATLAB实现传统谱减法以及两种改进的谱减法(增益函数谱减法、多带谱减法)的语音增强
  • 同一无线网络下的设备IP地址是否相同?
  • 第2周 PINN核心技术揭秘: 如何用神经网络求解偏微分方程
  • 【C语言】习题练手套餐 2
  • [项目总结] 基于Docker与Nginx对项目进行部署
  • 神经正切核推导(2)
  • Python模型优化技巧
  • Redis 面试场景
  • MySQL 索引失效及其解决办法
  • Ctrl+鼠标滚动阻止页面放大/缩小
  • 开发积累总结
  • C++虚函数与类对象模型深度解析
  • 3d世界坐标系转屏幕坐标系
  • 【2025】基于Springboot + vue + 协同过滤算法实现的旅游推荐系统
  • AI数据治理破局的战略重构
  • QT6安装与概念介绍
  • Python包管理工具uv 国内源配置
  • ABP VNext + Webhook:订阅与异步回调
  • Docker(二):开机自启动与基础配置、镜像加速器优化与疑难排查指南
  • Lua基础语法
  • 2025年渗透测试面试题总结-匿名[实习]安全工程师(安全厂商)(题目+回答)