RabbitMQ入門及常用的5種模式的簡單使用（一）
RabbitMQ是一個非常常用也非常強大的訊息中介軟體，主要用於應用與應用之間的通訊，有五種常見的使用方式，分別是：簡單模式，工作模式，釋出訂閱模式，路由模式以及萬用字元模式，這裡主要是簡單模式和工作模式！
package cn.itcast.rabbitmq.simple;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* rabbitmq的第一種模式：簡單模式（點對點）
* 該模式下，一個生產者對應一個消費者
*
* @author 桑偉東
*
*/
public class Send {
private final static String QUEUE_NAME = "test_queue";
public static void main(String[] argv) throws Exception {
// 獲取到連線以及mq通道
Connection connection = ConnectionUtil.getConnection();
// 從連線中建立通道
Channel channel = connection.createChannel();
// 宣告（建立）佇列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 訊息內容
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
// 關閉通道和連線
channel.close();
connection.close();
}
}
為了簡化程式碼，這裡抽取了一個公共的方法，用來獲取連線
package cn.itcast.rabbitmq.util;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
/**
* 通用的獲取連線的工具類
* @author 桑偉東
*
*/
public class ConnectionUtil {
public static Connection getConnection() throws Exception {
//定義連線工廠
ConnectionFactory factory = new ConnectionFactory();
//設定服務地址
factory.setHost("localhost");
//埠
factory.setPort(5672);
//設定賬號資訊，使用者名稱、密碼、vhost
factory.setVirtualHost("/swd");//虛擬主機
factory.setUsername("root");
factory.setPassword("root");
// 通過工程獲取連線
Connection connection = factory.newConnection();
return connection;
}
}
package cn.itcast.rabbitmq.simple;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
/**
* 簡單模式的接收訊息
* @author 桑偉東
*
*/
public class Recv {
private final static String QUEUE_NAME = "test_queue";
public static void main(String[] argv) throws Exception {
// 獲取到連線以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 宣告佇列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 定義佇列的消費者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 監聽佇列
channel.basicConsume(QUEUE_NAME, true, consumer);
// 獲取訊息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
}
}
}
工作模式
package cn.itcast.rabbitmq.work;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Send {
private final static String QUEUE_NAME = "test_queue_work";
public static void main(String[] argv) throws Exception {
// 獲取到連線以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 宣告佇列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
for (int i = 0; i < 50; i++) {
// 訊息內容
String message = "" + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
Thread.sleep(i * 10);
}
channel.close();
connection.close();
}
}
package cn.itcast.rabbitmq.work;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv {
private final static String QUEUE_NAME = "test_queue_work";
public static void main(String[] argv) throws Exception {
// 獲取到連線以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 宣告佇列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 同一時刻伺服器只會發一條訊息給消費者
channel.basicQos(1);
// 定義佇列的消費者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 監聽佇列，手動返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);
// 獲取訊息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
//休眠
Thread.sleep(10);
// 返回確認狀態
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}
package cn.itcast.rabbitmq.work;
import cn.itcast.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
public class Recv2 {
private final static String QUEUE_NAME = "test_queue_work";
public static void main(String[] argv) throws Exception {
// 獲取到連線以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 宣告佇列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 同一時刻伺服器只會發一條訊息給消費者
channel.basicQos(1);
// 定義佇列的消費者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 監聽佇列，手動返回完成狀態
channel.basicConsume(QUEUE_NAME, false, consumer);
// 獲取訊息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
// 休眠1秒
Thread.sleep(1000);
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}