当前位置：首页 > news >正文

rabbitmq的使用介绍

news 来源：原创 2025/5/26 5:32:05

一.队列工作模式介绍

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;
}