java-rabbitmq-官網執行個體02
描述:
1.定義持久化隊列,發送持久化消息,消息接受者需要手動應答,MQ才會删除隊列中的消息
2.使用 channel.basicQos() 設定多個消費者時,消息将平均分發
運作:
D2_NewTask.main(); //生産者
D2_Worker.main(); //消費者
package com.example.tutorials;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import java.util.Scanner;
/**
* 1.定義持久化隊列,發送持久化消息,消息接受者需要手動應答,MQ才會删除隊列中的消息
* 2.使用 channel.basicQos() 設定多個消費者時,消息将平均分發
* @create 2017-08-29
* amqp-client 4.2.0
**/
public class D2_NewTask {
private static final String TASK_QUEUE_NAME = "task_queue";
/**
* 生産者, Work queues
* @param argv
* @throws Exception
*/
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
//設定登入賬号
factory.setHost(ServerInfo.host);
factory.setPort(ServerInfo.port);
factory.setUsername(ServerInfo.uName);
factory.setPassword(ServerInfo.uPwd);
//連結伺服器
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//定義一個隊列
boolean duiable=true;//隊列持久化
boolean exclusive = false;//排他隊列
boolean autoDelete=false;//沒有consumer時,隊列是否自動删除
channel.queueDeclare(TASK_QUEUE_NAME, duiable, exclusive, autoDelete, null);
//發送消息
System.out.println("輸入要發送的消息,退出輸入 x ");
String message = "";
do{
Scanner scanner = new Scanner(System.in);
message = scanner.next();
channel.basicPublish("", TASK_QUEUE_NAME,
MessageProperties.PERSISTENT_TEXT_PLAIN,//設定消息持久化
message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "'");
}while(!"x".equals(message));
channel.close();
connection.close();
}
}
package com.example.tutorials;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
* 1.定義持久化隊列,發送持久化消息,消息接受者需要手動應答,MQ才會删除隊列中的消息
* 2.使用 channel.basicQos() 設定多個消費者時,消息将平均分發
* @create 2017-08-29
* amqp-client 4.2.0
**/
public class D2_Worker {
private static final String TASK_QUEUE_NAME = "task_queue";
/**
* 消費者, Work queues
* @param argv
* @throws Exception
*/
public static void main(String[] argv) {
//多個消費者,指定了相同名稱的隊列,向隊列發送消失時,消息将推送給随機一個消費者消費。可以 調用channel.basicQos(1) 設定後,消息将以輪詢的形式推送到每個消費者
//啟動兩個線程,模拟消費者
new Thread(new Runnable() {
@Override
public void run() {
try {
startConsumer("consumer_01");
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
startConsumer("consumer_02");
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
/**
* 啟動一個消費者
* @param consumerName
* @throws Exception
*/
public static void startConsumer(final String consumerName) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
//設定登入賬号
factory.setHost(ServerInfo.host);
factory.setPort(ServerInfo.port);
factory.setUsername(ServerInfo.uName);
factory.setPassword(ServerInfo.uPwd);
//連結伺服器
final Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
//定義一個隊列
boolean duiable=true;//持久化
boolean exclusive = false;//排他隊列
boolean autoDelete=false;//沒有consumer時,隊列是否自動删除
channel.queueDeclare(TASK_QUEUE_NAME, duiable, exclusive, autoDelete, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
//平均分發消息,有多個消費者時,消息将以輪詢方式接收到消息
channel.basicQos(1);
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleConsumeOk(String consumerTag) {
super.handleConsumeOk(consumerTag);
System.out.println(" ["+consumerName+"] 已經注冊成功! ");
System.out.println("----------------------------------------------");
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("----------------------------------------------");
System.out.println(" ["+consumerName+"] Received '" + message + "'");
try {
doWork(message);//消費消息
} finally {
System.out.println(" ["+consumerName+"] Done");
//手動發送應答,使消息在隊列中被删除
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
//啟動消費者,接收消息
boolean autoAck=false;//自動應答,true=自動發送應答;false=手動發送應答;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
}
/**
* 模拟延時消費消息
* @param task
*/
private static void doWork(String task) {
for (char ch : task.toCharArray()) {
if (ch == '.') {
try {
Thread.sleep(1000);
} catch (InterruptedException _ignored) {
Thread.currentThread().interrupt();
}
}
}
}
}