RabbitMQ 工作队列模式(Work queues)

1介绍

1.1工作模式和简单模式的区别

durable（是否持久化）：

这个参数改成true,表示需要持久化。当然如果rabbitmq接受到消息但还没有保存的时候，突然挂了，这样消息仍然会丢失。（sender 和 receiver 一样配置）

接受者：

autoAck（是否自动应答）：

应答之后，rabbitmq就会删除消息

channel .basicConsume ( QUEUE_NAME , autoAck , consumer );

channel . basicQos ( 1 );

设置每次取用的消息数量，每次只取一条消息

2 代码

2.1 生产者，持续发送消息，每隔1秒发送一条消息，共50条

package com.yzcq.rabbitmq.workqueue ;

import com.rabbitmq.client.Channel ;

import com.rabbitmq.client.Connection ;

import com.rabbitmq.client.MessageProperties ;

import com.yzcq.rabbitmq.util.ConnectionUtil ;

import java.io.IOException ;

import java.util.concurrent.TimeoutException ;

/**

* 生产者

*/

public class Send {

private static final String QUEUE_NAME = "task_queue" ;

public static void main ( String [] args) throws IOException , TimeoutException {

// 获取连接

Connection connection = ConnectionUtil . getConnection ();

// 创建通道

Channel channel = connection . createChannel ();

// 持久化

boolean durable = true ;

// 声明一个持久化的 队列

channel . queueDeclare ( QUEUE_NAME , durable , false , false , null );

// 循环发消息发消息

for ( int i = 0 ; i < 50 ; i ++) {

String msg = "hello,I am task_queue [" + i + "]" ;

channel . basicPublish ( "" , QUEUE_NAME , MessageProperties . PERSISTENT_TEXT_PLAIN , msg . getBytes ());

System . out . println ( "----send msg: " + msg );

try {

Thread . sleep ( 1000 );

}

catch ( InterruptedException e) {

Thread . currentThread (). interrupt ();

}

}

channel . close ();

connection . close ();

}

}

2.2 消费者1，每隔1秒钟，接受1条消息

package com.yzcq.rabbitmq.workqueue ;

import com.rabbitmq.client.AMQP . BasicProperties ;

import com.rabbitmq.client.Channel ;

import com.rabbitmq.client.Connection ;

import com.rabbitmq.client.Consumer ;

import com.rabbitmq.client.DefaultConsumer ;

import com.rabbitmq.client.Envelope ;

import com.yzcq.rabbitmq.util.ConnectionUtil ;

import java.io.IOException ;

import java.util.concurrent.TimeoutException ;

/**

* 工作队列

*/

public class Receiver1 {

private static final String QUEUE_NAME = "task_queue" ;

public static void main ( String [] args) throws IOException , TimeoutException {

final Connection connection = ConnectionUtil . getConnection ();

final Channel channel = connection . createChannel ();

// 声明我要使用的队列， QUEUE_NAME 是此队列的唯一标志

channel . queueDeclare ( QUEUE_NAME , true , false , false , null );

// 设置每次取用的消息数量，每次只取一条消息

channel . basicQos ( 1 );

final Consumer consumer = new DefaultConsumer ( channel ) {

@Override

public void handleDelivery ( String consumerTag, Envelope envelope, BasicProperties properties, byte [] body) throws IOException {

String message = new String (body, "UTF-8" );

System . out . println ( " [1] Received '" + message + "'" );

try {

Thread . sleep ( 1000 );

}

catch ( InterruptedException _ignored) {

Thread . currentThread (). interrupt ();

}

finally {

System . out . println ( " [1] DeliveryTag : " + envelope. getDeliveryTag ());

channel . basicAck (envelope. getDeliveryTag (), false );

}

}

};

// 是否自动应答，应答之后， rabbitmq 就会删除消息

boolean autoAck = false ;

// 设置消费

channel . basicConsume ( QUEUE_NAME , autoAck , consumer );

}

}

2.2 消费者2，每隔2秒钟，接受1条消息

package com.yzcq.rabbitmq.workqueue ;

import com.rabbitmq.client.AMQP . BasicProperties ;

import com.rabbitmq.client.Channel ;

import com.rabbitmq.client.Connection ;

import com.rabbitmq.client.Consumer ;

import com.rabbitmq.client.DefaultConsumer ;

import com.rabbitmq.client.Envelope ;

import com.yzcq.rabbitmq.util.ConnectionUtil ;

import java.io.IOException ;

import java.util.concurrent.TimeoutException ;

/**

* 工作队列

*/

public class Receiver2 {

private static final String QUEUE_NAME = "task_queue" ;

public static void main ( String [] args) throws IOException , TimeoutException {

final Connection connection = ConnectionUtil . getConnection ();

final Channel channel = connection . createChannel ();

// 声明我要使用的队列， QUEUE_NAME 是此队列的唯一标志

channel . queueDeclare ( QUEUE_NAME , true , false , false , null );

// 设置每次取用的消息数量，每次只取一条消息

channel . basicQos ( 1 );

final Consumer consumer = new DefaultConsumer ( channel ) {

@Override

public void handleDelivery ( String consumerTag, Envelope envelope, BasicProperties properties, byte [] body) throws IOException {

String message = new String (body, "UTF-8" );

System . out . println ( " [2] Received '" + message + "'" );

try {

Thread . sleep ( 2000 );

}

catch ( InterruptedException _ignored) {

Thread . currentThread (). interrupt ();

}

finally {

System . out . println ( " [2] DeliveryTag : " + envelope. getDeliveryTag ());

channel . basicAck (envelope. getDeliveryTag (), false );

}

}

};

// 是否自动应答，应答之后， rabbitmq 就会删除消息

boolean autoAck = false ;

// 设置消费

channel . basicConsume ( QUEUE_NAME , autoAck , consumer );

}

}