RabbitMQ（二）：Java-工作队列模式

RabbitMQ系列二

文章目录

工作队列模式(work queue)引入pom生产者消费者1消费者2

---

工作队列模式(work queue)

一个消息轮流被接收

特点：

一个生产者发送消息多个消费者共同监听一个队列的消息消息不能被重复消费采用轮询的方式将消息平均发送给消费者(c1,c2,c1,c2…交替)

引入pom

<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> </dependency>

生产者

public class Send { // 队列名 private final static String QUEUE_NAME = "FIRST-MQ"; public static void main(String[] args) throws IOException, TimeoutException { // 建立新的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("106.**.**.82"); factory.setPort(5672); // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = null; Channel channel = null; try{ connection = factory.newConnection(); // 创建内部会话通道，生产者和mq服务通信都在channel中完成 channel = connection.createChannel(); // 声明交换机，这里暂时没有,就会用默认的交换机 // 声明创建队列，如果没有，则创建 // 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments /** * queue：队列名称 * durable：是否持久化，存到磁盘中 * exclusive:是否排他，是否独占连接，该队列只允许在这个connection中访问，如果关闭连接，则删除队列 * autoDelete:不再使用时自动删除。和exclusive 一同设为true 可以实现临时队列。 * args：扩展参数，比如存活时间 */ channel.queueDeclare(QUEUE_NAME,false,false,false, null); String message = "work queue 模式，消息来了。"; // 发送消息 // 参数：String exchange, String routingKey, BasicProperties props, byte[] body /** * exchange:交换机 如果不指定("")，就默认交换机 * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称 * props:额外属性 * body：消息内容 */ channel.basicPublish("",QUEUE_NAME,null,message.getBytes()); System.out.println("send message:"+message); } catch (TimeoutException | IOException e) { e.printStackTrace(); } finally { // 先关闭通道 channel.close(); // 再关闭连接 connection.close(); } } }

消费者1

public class Recv { private final static String QUEUE_NAME = "FIRST-MQ"; public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("106.**.**.82"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); try{ Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] Waiting for messages."); // 定义实现消费方法 Consumer consumer = new DefaultConsumer(channel) { /** * 当接收到消息后，此方法将被调用 * @description: * @param consumerTag: (非必须)消费者标签：标识消费者，在监听队列的时候设置channel.basicConsume中设置 * @param envelope: 信封，通过envelope 可以获取很多信息，交换机、消息id(可用于消息接收ack)、路由Key等等 * @param properties:消息的属性，发送消息时 设置的额外属性 * @param body :消息内容 */ @SneakyThrows @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, StandardCharsets.UTF_8); System.out.println("消费者1 receive:" + message); TimeUnit.SECONDS.sleep(1); // 手动回复ack // channel.basicAck(envelope.getDeliveryTag(),false); } }; // 接收消息 监听队列 // 参数：String queue, boolean autoAck, Consumer callback /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */ channel.basicConsume(QUEUE_NAME, true, consumer); } catch (IOException | TimeoutException | InterruptedException e) { e.printStackTrace(); } } }

消费者2

public class Recv2 { private final static String QUEUE_NAME = "FIRST-MQ"; public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("106.**.**.82"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); try{ Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] Waiting for messages."); // 定义实现消费方法 Consumer consumer = new DefaultConsumer(channel) { /** * 当接收到消息后，此方法将被调用 * @description: * @param consumerTag: (非必须)消费者标签：标识消费者，在监听队列的时候设置channel.basicConsume中设置 * @param envelope: 信封，通过envelope 可以获取很多信息，交换机、消息id(可用于消息接收ack)、路由Key等等 * @param properties:消息的属性，发送消息时 设置的额外属性 * @param body :消息内容 */ @SneakyThrows @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, StandardCharsets.UTF_8); System.out.println("消费者2 receive:" + message); TimeUnit.SECONDS.sleep(1); // 手动回复ack // channel.basicAck(envelope.getDeliveryTag(),false); } }; // 接收消息 监听队列 // 参数：String queue, boolean autoAck, Consumer callback /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */ channel.basicConsume(QUEUE_NAME, true, consumer); } catch (IOException | TimeoutException | InterruptedException e) { e.printStackTrace(); } } }

先启动两个消费者，然后再多次启动生产者，就能看到效果啦。