简单队列模式:最简单的工作队列,其中一个生产者,一个消费者,一个队列,也称为点对点模式。
P:消息的生产者
C:消息的消费者
红色:队列
生产者将消息发送到队列(通过默认交换机),消费者从队列中获取消息。
1. pom.xml引入rabbitmq依赖
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.13.1</version> </dependency>
2. 生产者
public static void main(String[] args) { // 设置连接参数 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672); connectionFactory.setUsername("admin"); connectionFactory.setPassword("123456"); connectionFactory.setVirtualHost("/"); Connection connection = null; Channel channel = null; try { // 创建连接Connection connection = connectionFactory.newConnection("生产者"); // 创建channel channel = connection.createChannel(); String queueName = "simple-queue1"; //创建队列 // 队列名称,是否持久化,是否独占,是否自动删除(当最后一条消息消费完毕后自动删除),附加属性 channel.queueDeclare(queueName, true, false, false, null); // 发送消息到队列 // exchange,routingKey, props,message body channel.basicPublish("", queueName, null, "hello message".getBytes(StandardCharsets.UTF_8)); System.out.println("消息发送成功"); } catch (IOException | TimeoutException e) { e.printStackTrace(); } finally { // 关闭channel和connection if (channel != null && channel.isOpen()) { try { channel.close(); } catch (IOException | TimeoutException e) { e.printStackTrace(); } } if (connection != null && connection.isOpen()) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } } }
3. 消费者
public class Consumer { public static void main(String[] args) { // 设置连接参数 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672); connectionFactory.setUsername("admin"); connectionFactory.setPassword("123456"); connectionFactory.setVirtualHost("/"); Connection connection = null; Channel channel = null; try { // 创建连接Connection connection = connectionFactory.newConnection("消费者"); // 创建channel channel = connection.createChannel(); // 监听发送来的消息 // 队列名称,是否自动确认消息,接收消息的回调,消费者取消时的回调 channel.basicConsume("simple-queue1", true, (consumerTag, message) -> System.out.println("收到消息:" + new String(message.getBody(), StandardCharsets.UTF_8)), consumerTag -> System.out.println("消费者取消订阅")); while (true) { //阻塞进程, 防止消费者程序退出 } } catch (IOException | TimeoutException e) { e.printStackTrace(); } finally { // 关闭channel和connection if (channel != null && channel.isOpen()) { try { channel.close(); } catch (IOException | TimeoutException e) { e.printStackTrace(); } } if (connection != null && connection.isOpen()) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } } }