- RabbitMQ 教程
- RabbitMQ - 主页
- RabbitMQ - 概述
- RabbitMQ - 环境设置
- RabbitMQ - 功能
- RabbitMQ - 安装
- 基于队列的示例
- RabbitMQ - 生产者应用程序
- RabbitMQ - 消费者应用程序
- RabbitMQ - 测试应用程序
- 基于主题的示例
- RabbitMQ - 发布者应用程序
- RabbitMQ - 订阅者应用程序
- RabbitMQ - 测试应用程序
- RabbitMQ 有用资源
- RabbitMQ - 快速指南
- RabbitMQ - 有用的资源
- RabbitMQ - 讨论
RabbitMQ - 消费者应用程序
现在让我们创建一个消费者应用程序,它将从 RabbitMQ 队列接收消息。
创建项目
使用 eclipse,选择File → New → Maven Project。勾选“创建一个简单项目(跳过原型选择)”,然后单击“下一步”。
输入详细信息,如下图所示:
groupId - com.tutorialspoint
artifactId - 消费者
版本- 0.0.1-SNAPSHOT
名称- RabbitMQ 消费者
单击完成按钮,将创建一个新项目。
pom.xml
现在更新 pom.xml 的内容以包含 ActiveMQ 的依赖项。
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.tutorialspoint.activemq</groupId> <artifactId>consumer</artifactId> <version>0.0.1-SNAPSHOT</version> <name>RabbitMQ Consumer</name> <properties> <java.version>11</java.version> </properties> <dependencies> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.14.2</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.26</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.26</version> </dependency> </dependencies> </project>
现在创建一个 Consumer 类,它将从 RabbitMQ 队列接收消息。
package com.tutorialspoint.rabbitmq; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; public class Consumer { private static String QUEUE = "MyFirstQueue"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE, false, false, false, null); System.out.println("Waiting for messages. To exit press CTRL+C"); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), StandardCharsets.UTF_8); System.out.println("Received '" + message + "'"); }; channel.basicConsume(QUEUE, true, deliverCallback, consumerTag -> { }); } }
消费者类创建一个连接,创建一个通道,如果不存在则创建一个队列,然后从队列接收消息(如果有),并且它将继续轮询队列中的消息。一旦消息被传递,它就会由 basicConsume() 方法使用 DeliverCallback 进行处理。
我们将在RabbitMQ - 测试应用程序一章中运行此应用程序。