热门标签
热门文章
- 1leetcode 523.连续的子数组的和——前缀和+哈希表_前缀和 hash leetcode
- 22021年中国物流仓储系统集成商竞争力排行TOP20_京东、快仓、华晓、嘉腾
- 3【基于react的立体柱状图的实现】_react 立体柱状图
- 4Python编程基础教程：量化交易入门_量化交易入门 csdn
- 5数据结构课程设计--通讯录管理（C语言实现）_c语言数据结构顺序表-通讯录管理
- 6RabbitMQ学习笔记（五）——RabbitMQ集群搭建&入门_mq的搭建和底层配置
- 7如何进行PHP中的数据库操作？_php 操作数据库
- 8Python词性分类 NLTK介绍 安装NLTK NLTK词性分类示例 spaCy介绍 安装spaCy spaCy词性分类示例——《跟老吕学Python编程》附录资料
- 9chrome调用linux本地程序,谷歌宣布Chromebook很快将能够运行本地Linux应用程序
- 10AI智能写作哪个好用？推荐4款一键生成的AI写作软件
当前位置: article > 正文
RabbitMQ-----发送接收
作者：很楠不爱3 | 2024-05-19 01:33:59
赞
踩
RabbitMQ-----发送接收
以下的例子都是没有创建交换机的，直接与队列进行信息接收，
实际开发不会。
在RabbitMQ中，交换机（Exchange）是用于接收生产者发送的消息，并根据一定的规则将消息路由到一个或多个队列（Queue）的
1。利用官方文档进行创建
单向发送
一个生产者向一个消息队列发送消息，一个消费者接收
我是创建的maven
先导入依赖
</dependency>
<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.21.0</version>
</dependency>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
生产者：
public class Send { //创建队列的名字 private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception { //创建连接工厂 ConnectionFactory是RabbitMQ Java客户端库中用于建立连接的工厂类。 ConnectionFactory factory = new ConnectionFactory(); //配置连接工厂以连接到本地主机（localhost）上的RabbitMQ服务器 factory.setHost("localhost"); //创建连接和通道 通道是发送和接收消息的主要接口-----(相当于jdbc客户端） //程序通过channel来操控rabbitmq try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { //通过通道调用queueDeclare方法来声明一个队列。 //这里队列的名称是前面定义的QUEUE_NAME。 //其他参数（是否持久化,重启后消息是否丢失、是否独占，是否只有当前只有创建当前消息队列的操作、 //是否自动删除、额外参数） channel.queueDeclare(QUEUE_NAME, false, false, false, null); String message = "Hello World!"; //通过通道调用basicPublish方法来发布消息。 //这里交换机名称是空字符串（""），表示不使用交换机而直接将消息发送到队列。 //路由键是队列的名称（QUEUE_NAME），因为没有使用交换机进行路由。 //消息的body是字符串message的UTF-8编码字节。 channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent '" + message + "'"); } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
点击运行后发现：
消费者：
public class Recv { private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); //新建一个频道，相当于客户端，操作消息队列的 Channel channel = connection.createChannel(); //建立连接 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); //定义了一个Lambda表达式作为消息交付的回调函数deliverCallback。 //当消息到达时，该函数会被调用，并将消息的内容转换为字符串并打印到控制台上。 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), StandardCharsets.UTF_8); System.out.println(" [x] Received '" + message + "."); }; //通过通道channel调用basicConsume方法来开始从队列中消费消息。这里指定了队列名称（QUEUE_NAME） //并设置autoAck参数为true，表示消息在被消费者接收后会立即被确认（即自动应答）。 //deliverCallback作为消息交付的回调函数， //一个空的Lambda表达式作为消费者标签的回调函数 channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
多消费者
生产者：
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.MessageProperties; import java.util.Scanner; public class NewTask { private static final String TASK_QUEUE_NAME = "mutiple-queue"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { //第二个参数表示队列是持久的（即使RabbitMQ服务器重启，队列也不会丢失） channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); //使用sacnner发送消息 Scanner scanner=new Scanner(System.in); while(scanner.hasNext()){ String message = scanner.nextLine(); // 发布消息 /* 交换器名称（这里使用空字符串，表示使用默认的交换器） 路由键（使用之前定义的常量TASK_QUEUE_NAME，这通常与队列名称相同，但也可以不同，取决于交换器的类型和配置） 消息属性（这里使用MessageProperties.PERSISTENT_TEXT_PLAIN，表示消息是持久的，并且是纯文本） 消息体（使用message.getBytes("UTF-8")将字符串转换为UTF-8编码的字节数组） */ channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8")); System.out.println(" [x] Sent '" + message + "'"); } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
消费者：
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; public class Worker { private static final String TASK_QUEUE_NAME = "mutiple-queue"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); final Connection connection = factory.newConnection(); for(int i=0;i<2;i++){ final Channel channel = connection.createChannel(); channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); //设置通道的基本服务质量（QoS），这里设置为1， // 意味着RabbitMQ一次只发送一个消息给消费者，直到该消息被确认 channel.basicQos(1); int t=i; //消息接收和处理: DeliverCallback deliverCallback = (consumerTag, delivery) -> { //当消息到达时，它首先会将消息体从字节转换为字符串。 String message = new String(delivery.getBody(), "UTF-8"); //输出一条消息，告知用户接收到了该消息。 System.out.println(" [x] Received '" +"编号"+t+ ":"+message + "'"); try { //调用doWork方法处理消息内容 doWork(message); } finally { //处理完成后，输出一条消息表示处理完成 System.out.println(" [x] Done"); //使用channel.basicAck方法确认消息已被处理，这样RabbitMQ就会从队列中删除该消息。 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } }; //消息确认机制，第二个参数false表示消息是手动确认的（即需要显式调用basicAck或basicNack） channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { }); } } /** * 遍历任务中的每个字符。 如果字符是.，则让当前线程休眠1秒（模拟耗时操作）。 如果在休眠期间被中断，则设置当前线程的中断状态。 * @param task */ private static void doWork(String task) { for (char ch : task.toCharArray()) { if (ch == '.') { try { //沉睡 Thread.sleep(3000); } catch (InterruptedException _ignored) { Thread.currentThread().interrupt(); } } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
2。Spring AMQP 提供的注解，用于简化 RabbitMQ 消息监听器的创建。
访问客户端的：
http://localhost:15672/
导入依赖：
首先配置MQ地址，在publisher服务的application.yml中添加配置：
spring:
rabbitmq:
host: 192.168.150.101 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: hmall # 用户名
password: 123 # 密码
- 1
- 2
- 3
- 4
- 5
- 6
- 7
上面是黑马提供的。但是我没有创建虚拟机而是本地，用的原始guest,如果你也一样，那就用我的
spring:
rabbitmq:
host: 127.0.0.1 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: /
username: guest # 用户名
password: guest # 密码
- 1
- 2
- 3
- 4
- 5
- 6
- 7
在RabbitMQ中，当消息生产者尝试发送消息到一个不存在的队列时，RabbitMQ服务器会自动为该生产者创建该队列，前提是生产者具有足够的权限来创建队列。（黑马是在客户端创建）
单向发送
1.在配置类里面加：
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
//不要偷懒 一定要声明
@Bean
public Queue chatMessageQueue(){
return new Queue("xx");
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
2.生产者，在测试类里面：
import org.junit.jupiter.api.Test; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; @SpringBootTest public class test { @Autowired private RabbitTemplate rabbitTemplate; @Test public void test() { // 队列名称 String queueName = "xx"; // 消息 String message = "hello, spring amqp!"; // 发送消息 rabbitTemplate.convertAndSend(queueName, message); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
3.消费者：
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SpringRabbitListener {
// 利用RabbitListener来声明要监听的队列信息
// 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
// 可以看到方法体中接收的就是消息体的内容
@RabbitListener(queues = "xx")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
System.out.println("spring 消费者接收到消息：【" + msg + "】");
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
多消费者WorkQueues模型
Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息
生产者
这次我们循环发送，模拟大量消息堆积现象。
在Test类中添加一个测试方法：
先在配置类声明：
@Bean
public Queue jjMessageQueue(){
return new Queue("jj");
}
- 1
- 2
- 3
- 4
/** * workQueue * 向队列中不停发送消息，模拟消息堆积。 */ @Test public void testWorkQueue() throws InterruptedException { // 队列名称 String queueName = "jj.queue"; // 消息 String message = "hello, message_"; for (int i = 0; i < 50; i++) { // 发送消息，每20毫秒发送一次，相当于每秒发送50条消息 rabbitTemplate.convertAndSend(queueName, message + i); Thread.sleep(20); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
消费者：
要模拟多个消费者绑定同一个队列，添加2个新的方法：
@RabbitListener(queues = "jj.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(20);
}
@RabbitListener(queues = "jj.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(200);
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
注意到这两消费者，都设置了Thead.sleep，模拟任务耗时：
- 消费者1 sleep了20毫秒，相当于每秒钟处理50个消息
- 消费者2 sleep了200毫秒，相当于每秒处理5个消息
3.3.3.测试
启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。
最终结果如下：
消费者1接收到消息：【hello, message_0】21:06:00.869555300 消费者2........接收到消息：【hello, message_1】21:06:00.884518 消费者1接收到消息：【hello, message_2】21:06:00.907454400 消费者1接收到消息：【hello, message_4】21:06:00.953332100 消费者1接收到消息：【hello, message_6】21:06:00.997867300 消费者1接收到消息：【hello, message_8】21:06:01.042178700 消费者2........接收到消息：【hello, message_3】21:06:01.086478800 消费者1接收到消息：【hello, message_10】21:06:01.087476600 消费者1接收到消息：【hello, message_12】21:06:01.132578300 消费者1接收到消息：【hello, message_14】21:06:01.175851200 消费者1接收到消息：【hello, message_16】21:06:01.218533400 消费者1接收到消息：【hello, message_18】21:06:01.261322900 消费者2........接收到消息：【hello, message_5】21:06:01.287003700 消费者1接收到消息：【hello, message_20】21:06:01.304412400 消费者1接收到消息：【hello, message_22】21:06:01.349950100 消费者1接收到消息：【hello, message_24】21:06:01.394533900 消费者1接收到消息：【hello, message_26】21:06:01.439876500 消费者1接收到消息：【hello, message_28】21:06:01.482937800 消费者2........接收到消息：【hello, message_7】21:06:01.488977100 消费者1接收到消息：【hello, message_30】21:06:01.526409300 消费者1接收到消息：【hello, message_32】21:06:01.572148 消费者1接收到消息：【hello, message_34】21:06:01.618264800 消费者1接收到消息：【hello, message_36】21:06:01.660780600 消费者2........接收到消息：【hello, message_9】21:06:01.689189300 消费者1接收到消息：【hello, message_38】21:06:01.705261 消费者1接收到消息：【hello, message_40】21:06:01.746927300 消费者1接收到消息：【hello, message_42】21:06:01.789835 消费者1接收到消息：【hello, message_44】21:06:01.834393100 消费者1接收到消息：【hello, message_46】21:06:01.875312100 消费者2........接收到消息：【hello, message_11】21:06:01.889969500 消费者1接收到消息：【hello, message_48】21:06:01.920702500 消费者2........接收到消息：【hello, message_13】21:06:02.090725900 消费者2........接收到消息：【hello, message_15】21:06:02.293060600 消费者2........接收到消息：【hello, message_17】21:06:02.493748 消费者2........接收到消息：【hello, message_19】21:06:02.696635100 消费者2........接收到消息：【hello, message_21】21:06:02.896809700 消费者2........接收到消息：【hello, message_23】21:06:03.099533400 消费者2........接收到消息：【hello, message_25】21:06:03.301446400 消费者2........接收到消息：【hello, message_27】21:06:03.504999100 消费者2........接收到消息：【hello, message_29】21:06:03.705702500 消费者2........接收到消息：【hello, message_31】21:06:03.906601200 消费者2........接收到消息：【hello, message_33】21:06:04.108118500 消费者2........接收到消息：【hello, message_35】21:06:04.308945400 消费者2........接收到消息：【hello, message_37】21:06:04.511547700 消费者2........接收到消息：【hello, message_39】21:06:04.714038400 消费者2........接收到消息：【hello, message_41】21:06:04.916192700 消费者2........接收到消息：【hello, message_43】21:06:05.116286400 消费者2........接收到消息：【hello, message_45】21:06:05.318055100 消费者2........接收到消息：【hello, message_47】21:06:05.520656400 消费者2........接收到消息：【hello, message_49】21:06:05.723106700
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
可以看到消费者1和消费者2竟然每人消费了25条消息：
- 消费者1很快完成了自己的25条消息
- 消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。
在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：
spring:
rabbitmq:
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
- 1
- 2
- 3
- 4
- 5
再次测试，发现结果如下：
消费者1接收到消息：【hello, message_0】21:12:51.659664200 消费者2........接收到消息：【hello, message_1】21:12:51.680610 消费者1接收到消息：【hello, message_2】21:12:51.703625 消费者1接收到消息：【hello, message_3】21:12:51.724330100 消费者1接收到消息：【hello, message_4】21:12:51.746651100 消费者1接收到消息：【hello, message_5】21:12:51.768401400 消费者1接收到消息：【hello, message_6】21:12:51.790511400 消费者1接收到消息：【hello, message_7】21:12:51.812559800 消费者1接收到消息：【hello, message_8】21:12:51.834500600 消费者1接收到消息：【hello, message_9】21:12:51.857438800 消费者1接收到消息：【hello, message_10】21:12:51.880379600 消费者2........接收到消息：【hello, message_11】21:12:51.899327100 消费者1接收到消息：【hello, message_12】21:12:51.922828400 消费者1接收到消息：【hello, message_13】21:12:51.945617400 消费者1接收到消息：【hello, message_14】21:12:51.968942500 消费者1接收到消息：【hello, message_15】21:12:51.992215400 消费者1接收到消息：【hello, message_16】21:12:52.013325600 消费者1接收到消息：【hello, message_17】21:12:52.035687100 消费者1接收到消息：【hello, message_18】21:12:52.058188 消费者1接收到消息：【hello, message_19】21:12:52.081208400 消费者2........接收到消息：【hello, message_20】21:12:52.103406200 消费者1接收到消息：【hello, message_21】21:12:52.123827300 消费者1接收到消息：【hello, message_22】21:12:52.146165100 消费者1接收到消息：【hello, message_23】21:12:52.168828300 消费者1接收到消息：【hello, message_24】21:12:52.191769500 消费者1接收到消息：【hello, message_25】21:12:52.214839100 消费者1接收到消息：【hello, message_26】21:12:52.238998700 消费者1接收到消息：【hello, message_27】21:12:52.259772600 消费者1接收到消息：【hello, message_28】21:12:52.284131800 消费者2........接收到消息：【hello, message_29】21:12:52.306190600 消费者1接收到消息：【hello, message_30】21:12:52.325315800 消费者1接收到消息：【hello, message_31】21:12:52.347012500 消费者1接收到消息：【hello, message_32】21:12:52.368508600 消费者1接收到消息：【hello, message_33】21:12:52.391785100 消费者1接收到消息：【hello, message_34】21:12:52.416383800 消费者1接收到消息：【hello, message_35】21:12:52.439019 消费者1接收到消息：【hello, message_36】21:12:52.461733900 消费者1接收到消息：【hello, message_37】21:12:52.485990 消费者1接收到消息：【hello, message_38】21:12:52.509219900 消费者2........接收到消息：【hello, message_39】21:12:52.523683400 消费者1接收到消息：【hello, message_40】21:12:52.547412100 消费者1接收到消息：【hello, message_41】21:12:52.571191800 消费者1接收到消息：【hello, message_42】21:12:52.593024600 消费者1接收到消息：【hello, message_43】21:12:52.616731800 消费者1接收到消息：【hello, message_44】21:12:52.640317 消费者1接收到消息：【hello, message_45】21:12:52.663111100 消费者1接收到消息：【hello, message_46】21:12:52.686727 消费者1接收到消息：【hello, message_47】21:12:52.709266500 消费者2........接收到消息：【hello, message_48】21:12:52.725884900 消费者1接收到消息：【hello, message_49】21:12:52.746299900
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
可以发现，由于消费者1处理速度较快，所以处理了更多的消息；消费者2处理速度较慢，只处理了6条消息。而最终总的执行耗时也在1秒左右，大大提升。
正所谓能者多劳，这样充分利用了每一个消费者的处理能力，可以有效避免消息积压问题。
本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/很楠不爱3/article/detail/590795
推荐阅读
相关标签
