中间件简单题目教学

题目1环境搭建与简单模式使用 Docker 启动 RabbitMQ 4.x 容器用户guest密码123456映射管理端口15672。编写 Java 原生生产者向队列test_queue发送消息Hello Exam。编写 Java 原生消费者消费该队列中的消息并打印到控制台使用手动确认模式。第一步启动 RabbitMQDocker 方式打开终端Windows 用 cmd 或 PowerShellMac/Linux 用终端输入以下命令bashdocker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USERguest -e RABBITMQ_DEFAULT_PASS123456 rabbitmq:4.0-management如果容器已存在先删除旧容器docker rm -f rabbitmq再执行上面命令。验证浏览器访问http://localhost:15672输入用户名guest密码123456看到管理界面即成功。提示考试时如果忘记命令可以上网搜索“docker run rabbitmq 4 management”。第二步创建 Maven 项目并添加依赖打开 IDEA点击File→New→Project。选择Maven不勾选Create from archetype点击Next。填写Namesimple-demoGroupIdcom.exampleArtifactIdsimple-demo点击Finish。等待 IDEA 加载完成打开pom.xml在dependencies标签内添加xmldependencies !-- RabbitMQ Java Client -- dependency groupIdcom.rabbitmq/groupId artifactIdamqp-client/artifactId version5.20.0/version /dependency /dependencies点击右下角弹出的Load Maven Changes或右键pom.xml→Maven→Reload project。第三步编写生产者Producer.java在src/main/java下创建包com.example然后创建类Producer复制以下代码javapackage com.example; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Producer { public static void main(String[] args) throws Exception { // 1. 创建连接工厂 ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(123456); factory.setVirtualHost(/); // 2. 创建连接和通道 try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 3. 声明队列持久化、非独占、不自动删除 channel.queueDeclare(test_queue, true, false, false, null); // 4. 发送消息 String message Hello Exam; channel.basicPublish(, test_queue, null, message.getBytes()); System.out.println(✅ 消息已发送: message); } } }第四步编写消费者Consumer.java在同一个包下创建Consumer类javapackage com.example; import com.rabbitmq.client.*; public class Consumer { public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(123456); Connection connection factory.newConnection(); Channel channel connection.createChannel(); // 声明同一个队列确保存在 channel.queueDeclare(test_queue, true, false, false, null); // 设置每次只取一条消息手动确认模式需要 channel.basicQos(1); System.out.println(⏳ 等待消息...); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { String message new String(body); System.out.println( 收到消息: message); // 手动确认 try { channel.basicAck(envelope.getDeliveryTag(), false); } catch (Exception e) { e.printStackTrace(); } } }; // 开始消费autoAck false channel.basicConsume(test_queue, false, consumer); } }第五步运行验证先运行Consumer右键 → Run控制台显示“等待消息...”。再运行Producer控制台显示“消息已发送”。观察Consumer控制台应打印“收到消息: Hello Exam”。如果一切正常说明简单模式成功。考试注意如果消费者没有打印检查是否先启动了消费者。如果报AuthenticationFailureException确认 Docker 启动时的密码是123456代码中密码一致。题目2工作队列模式一个生产者发送 20 条任务消息Task1 ~ Task20。两个消费者共同处理每个消费者每次只能取一条消息处理完成后才能取下一条。每个任务模拟处理 500ms打印“已完成任务名”。解题思路工作队列模式与简单模式代码几乎一样区别在于启动两个消费者实例。消费者端需设置channel.basicQos(1)实现公平分发。使用手动确认autoAckfalse确保消息处理完才删除。详细步骤复用上一个项目或者新建 Maven 项目依赖相同。生产者代码WorkQueueProducer.javajavapackage com.example; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class WorkQueueProducer { private static final String QUEUE_NAME work_queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(123456); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { channel.queueDeclare(QUEUE_NAME, true, false, false, null); for (int i 1; i 20; i) { String message Task i; channel.basicPublish(, QUEUE_NAME, null, message.getBytes()); System.out.println(✅ 发送: message); Thread.sleep(200); // 间隔发送 } } } }消费者代码WorkQueueConsumer.javajavapackage com.example; import com.rabbitmq.client.*; public class WorkQueueConsumer { private static final String QUEUE_NAME work_queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(123456); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 关键每次只取一条消息 channel.basicQos(1); System.out.println(等待任务...); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { String task new String(body); System.out.println( 收到: task); try { Thread.sleep(500); // 模拟处理耗时 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println(✅ 已完成: task); try { channel.basicAck(envelope.getDeliveryTag(), false); } catch (Exception e) { e.printStackTrace(); } } }; channel.basicConsume(QUEUE_NAME, false, consumer); } }启动两个消费者第一次运行WorkQueueConsumer右键 → Run。第二次运行WorkQueueConsumer需要允许多个实例并行。如果弹出提示“is not allowed to run in parallel”则修改运行配置点击顶部Run→Edit Configurations→ 找到WorkQueueConsumer→ 勾选Allow multiple instances或Modify options→Allow multiple instances→ OK。然后再运行第二次。运行生产者发送 20 条任务。观察两个消费者的控制台会交替打印任务或按处理速度公平分发每个任务只被一个消费者处理。如果某个消费者处理快它会多拿任务吗设置basicQos(1)后每个消费者同时只能有一个未确认的消息所以会轮询或公平分发不会出现一个累死一个闲死。题目3路由模式实现日志系统使用 Direct 交换机。生产者发送info、warning、error三种级别日志。消费者 A 只接收error日志。消费者 B 接收info和warning日志。验证消息路由正确。详细步骤创建 Maven 项目依赖同前。生产者DirectProducer.javajavapackage com.example; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class DirectProducer { private static final String EXCHANGE_NAME direct_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(123456); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明 direct 交换机 channel.exchangeDeclare(EXCHANGE_NAME, direct); // 发送三个级别的日志 String[] levels {info, warning, error}; for (String level : levels) { String message 这是 level 日志; channel.basicPublish(EXCHANGE_NAME, level, null, message.getBytes()); System.out.println(✅ 发送 [ level ]: message); } } } }消费者 A只收 errorjavapackage com.example; import com.rabbitmq.client.*; public class ConsumerA { private static final String EXCHANGE_NAME direct_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(123456); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, direct); // 创建临时队列 String queueName channel.queueDeclare().getQueue(); // 只绑定 error 路由键 channel.queueBind(queueName, EXCHANGE_NAME, error); System.out.println(消费者A 等待 error 日志...); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { String msg new String(body); System.out.println(消费者A 收到: msg); } }; channel.basicConsume(queueName, true, consumer); } }消费者 B收 info 和 warningjavapackage com.example; import com.rabbitmq.client.*; public class ConsumerB { private static final String EXCHANGE_NAME direct_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(123456); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, direct); String queueName channel.queueDeclare().getQueue(); // 绑定两个路由键 channel.queueBind(queueName, EXCHANGE_NAME, info); channel.queueBind(queueName, EXCHANGE_NAME, warning); System.out.println(消费者B 等待 info/warning 日志...); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { String msg new String(body); System.out.println(消费者B 收到: msg); } }; channel.basicConsume(queueName, true, consumer); } }启动顺序先启动ConsumerA和ConsumerB两个都启动注意允许多个实例。再启动DirectProducer。预期结果消费者A 只打印包含error的消息。消费者B 打印包含info和warning的消息。info和warning不会出现在 A 中error不会出现在 B 中。注意如果队列是临时队列每次运行消费者会生成不同队列名但绑定关系在每次启动时都会重新建立不影响结果。题目4Spring Boot 整合 RabbitMQ使用 Spring Boot 2.7.x 创建项目。提供一个 REST 接口POST /send接收参数message将消息发送到队列my_queue。监听器监听my_queue收到消息后打印到控制台并手动确认。配置文件中设置生产者确认和消费者手动确认。第一步创建 Spring Boot 项目IDEA 社区版手动方式由于社区版没有 Spring Initializr我们手动创建 Maven 项目并添加依赖。File→New→Project→ Maven → Next。填写Namespringboot-rabbit-demoGroupIdcom.exampleArtifactIdspringboot-rabbit-demo。修改pom.xmlxml?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdcom.example/groupId artifactIdspringboot-rabbit-demo/artifactId version1.0-SNAPSHOT/version parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version2.7.18/version relativePath/ /parent properties java.version11/java.version /properties dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId /plugin /plugins /build /project创建src/main/resources目录右键src/main→ New → Directory →resources然后右键resources→ Mark Directory as → Resources Root。在resources下创建application.ymlyamlserver: port: 8080 spring: rabbitmq: host: localhost port: 5672 username: guest password: 123456 virtual-host: / publisher-confirm-type: correlated # 生产者确认 listener: simple: acknowledge-mode: manual # 消费者手动确认第二步编写启动类在src/main/java下创建包com.example.demo创建Application.javajavapackage com.example.demo; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; SpringBootApplication public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } }第三步编写生产者服务在包下创建service子包新建SendService.javajavapackage com.example.demo.service; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; Service public class SendService { Autowired private RabbitTemplate rabbitTemplate; private static final String QUEUE_NAME my_queue; public void sendMessage(String message) { rabbitTemplate.convertAndSend(QUEUE_NAME, message); System.out.println(发送消息: message); } }第四步编写控制器在controller子包下新建SendController.javajavapackage com.example.demo.controller; import com.example.demo.service.SendService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; RestController public class SendController { Autowired private SendService sendService; PostMapping(/send) public String send(RequestParam String message) { sendService.sendMessage(message); return 消息已发送: message; } }第五步编写消费者监听器在listener子包下新建MessageListener.javajavapackage com.example.demo.listener; import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.messaging.handler.annotation.Header; import org.springframework.stereotype.Component; import java.io.IOException; Component public class MessageListener { RabbitListener(queues my_queue) public void handleMessage(String message, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) { System.out.println(收到消息: message); try { // 手动确认 channel.basicAck(deliveryTag, false); } catch (IOException e) { e.printStackTrace(); } } }注意队列my_queue需要存在。可以在启动类中增加一个Bean声明队列或者使用RabbitListener(queuesToDeclare Queue(my_queue))。为了简单我们手动声明一下。在Application类中添加javaBean public Queue myQueue() { return new Queue(my_queue, true); }第六步运行测试确保 RabbitMQ 已启动Docker。运行Application主类。使用 Postman 或浏览器需支持 POST发送请求URL:http://localhost:8080/send?messageHelloSpringBoot方法POST观察控制台生产者打印“发送消息...”消费者打印“收到消息...”。访问 RabbitMQ 管理界面my_queue队列中的消息应已被消费Ready0。考试提示如果忘记队列声明启动时可能会报错队列不存在。可以上网搜索“Spring Boot RabbitMQ 声明队列”。题目5微服务 RabbitMQ 综合题已有 Eureka Server 运行在localhost:8761考试时可能已经启动或需要你自己启动。编写订单微服务order-service注册到 Eureka端口 8081提供一个POST /order/create接口生成订单ID发送消息到order.exchange路由键order.create。编写支付微服务payment-service注册到 Eureka端口 8082监听order.create消息打印“支付处理订单ID xxx”。两者通过 RabbitMQ 解耦。第一步启动 Eureka Server如果考试未提供如果你需要自己写 Eureka Server按以下步骤创建 Maven 模块eureka-serverpom.xml添加xmldependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-netflix-eureka-server/artifactId /dependency启动类加EnableEurekaServer。application.ymlyamlserver: port: 8761 eureka: client: register-with-eureka: false fetch-registry: false运行访问http://localhost:8761确认。第二步创建订单服务order-service新建 Maven 模块order-service父工程为microservice-demo如果没有父工程可以单独建项目但依赖需要自己管理版本。pom.xml依赖xmldependencies dependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-netflix-eureka-client/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency /dependencies在父工程pom.xml的dependencyManagement中需添加 Spring Cloud 版本参考题目4中的父工程配置或者直接在每个子模块中指定版本但推荐用父工程管理。启动类OrderServiceApplication.javajavaSpringBootApplication EnableEurekaClient public class OrderServiceApplication { public static void main(String[] args) { SpringApplication.run(OrderServiceApplication.class, args); } Bean public DirectExchange orderExchange() { return new DirectExchange(order.exchange); } Bean public Queue orderQueue() { return new Queue(order.queue); } Bean public Binding binding() { return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(order.create); } }配置文件application.ymlyamlserver: port: 8081 spring: application: name: order-service rabbitmq: host: localhost port: 5672 username: guest password: 123456 eureka: client: service-url: defaultZone: http://localhost:8761/eureka/编写OrderControllerjavaRestController public class OrderController { Autowired private RabbitTemplate rabbitTemplate; PostMapping(/order/create) public String createOrder(RequestParam String productName) { String orderId UUID.randomUUID().toString(); String message 订单ID: orderId , 商品: productName; rabbitTemplate.convertAndSend(order.exchange, order.create, message); return 订单创建成功ID: orderId; } }第三步创建支付服务payment-service类似步骤创建模块payment-service端口 8082服务名payment-service。启动类无需额外声明队列但如果先启动支付服务队列不存在会报错可以使用RabbitListener(queuesToDeclare Queue(order.queue))自动创建队列。消费者监听器javaComponent public class PaymentListener { RabbitListener(queues order.queue) public void handleOrderMessage(String message) { System.out.println(支付服务收到消息: message); // 模拟支付处理 System.out.println(✅ 支付处理完成); } }配置文件application.yml除了端口和名字其他同 order-service。第四步启动与测试启动 RabbitMQDocker。启动 Eureka Server。启动 Payment Service让消费者先监听。启动 Order Service。发送 POST 请求http://localhost:8081/order/create?productName手机。观察 Order Service 控制台应该有发送日志和 Payment Service 控制台应该打印收到消息。常见问题如果支付服务启动时队列不存在可以在监听注解中使用queuesToDeclare Queue(order.queue)自动创建。如果消息发送失败检查 RabbitMQ 是否启动交换机和队列名称是否一致。