java rabbitmq 工具类_RabbitMq通用管理工具类
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.*;
import org.apache.log4j.Logger;
/**
* RabbitMq通用管理工具類
* RabbitMQ是AMQP(高級消息隊列協議)的標準實現
* 1.單發送單(多)接收模式；
* 2.fanout發布訂閱模式(fanout)；
* 3.routing路由模式(direct)；
* 4.topic通配符模式(topic)
* Broker:簡單來說就是消息隊列服務器實體;
Channel:消息通道，在客戶端的每個連接里，可建立多個channel，每個channel代表一個會話任務;
Exchange:交換機，決定了消息路由規則，它指定消息按什么規則，路由到哪個隊列;
Queue:消息隊列載體，每個消息都會被投入到一個或多個隊列;
Binding:綁定了Queue和Exchange，意即為符合什么樣路由規則的消息，將會放置入哪一個消息隊列;
Routing Key:路由關鍵字，exchange根據這個關鍵字進行消息投遞;
vhost:虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離;
producer:消息生產者，就是投遞消息的程序;
consumer:消息消費者，就是接受消息的程序;
* 消息隊列持久化包括3個部分:
(1)exchange持久化，在聲明時指定durable => 1
(2)queue持久化，在聲明時指定durable => 1
(3)消息持久化，在投遞時指定delivery_mode => 2(1是非持久化)
如果exchange和queue都是持久化的，那么它們之間的binding也是持久化的。如果exchange和queue兩者之間有一個持久化，一個非持久化，就不允許建立綁定。
* 1.將交換機置為可持久;
2.將通道置為可持久;
3.消息發送時設置可持久;
當我們"生產"了一條可持久化的消息，嘗試中斷MQ服務，啟動消費者獲取消息，消息依然能夠恢復。相反，則拋出異常;
* 消息隊列的使用過程大概如下:
(1)客戶端連接到消息隊列服務器，打開一個channel;
(2)客戶端聲明一個exchange，并設置相關屬性;
(3)客戶端聲明一個queue，并設置相關屬性;
(4)客戶端使用routing key，在exchange和queue之間建立好綁定關系;
(5)客戶端投遞消息到exchange, exchange接收到消息后，就根據消息的key和已經設置的binding，進行消息路由，將消息投遞到一個或多個隊列里
* exchange也有幾個類型，
完全根據key進行投遞的叫做Direct交換機，例如，綁定時設置了routing key為"abc"，那么客戶端提交的消息，只有設置了key為"abc"的才會投遞到隊列。
對key進行模式匹配后進行投遞的叫做Topic交換機，符號"#"匹配一個或多個詞，符號"*"匹配正好一個詞。例如"abc.#"匹配"abc.def.ghi"，"abc.*"只匹配"abc.def"。
還有一種不需要key的，叫做Fanout交換機，它采取廣播模式，一個消息進來時，投遞到與該交換機綁定的所有隊列。
* Exchange
Durability 持久性，這是exchange的可選屬性，如果你Durability設置為false，那些當前會話結束的時候，該exchange也會被銷毀;
Auto delete 當沒有隊列或者其他exchange綁定到此exchange的時候，該exchange被銷毀;
Internal 表示這個exchange不可以被client用來推送消息，僅用來進行exchange和exchange之間的綁定;
無法聲明2個名稱相同 但是類型卻不同的exchange;
* Queue
Durability 和exchange相同，未持久化的隊列，服務重啟后銷毀;
Auto delete 當沒有消費者連接到該隊列的時候，隊列自動銷毀;
Exclusive 使隊列成為私有隊列，只有當前應用程序可用，當你需要限制隊列只有一個消費者，這是很有用的;
擴展屬性如下對應源程序 RabbitMQ.Client.IModel.QueueDeclare(string, bool, bool, bool, System.Collections.Generic.IDictionary)最后的參數,
Message TTL 當一個消息被推送在該隊列的時候 可以存在的時間 單位為ms，(對應擴展參數argument "x-message-ttl" );
Auto expire 在隊列自動刪除之前可以保留多長時間(對應擴展參數argument "x-expires");
Max length 一個隊列可以容納的已準備消息的數量(對應擴展參數argument "x-max-length");
一旦創建了隊列和交換機，就不能修改其標志了。例如，如果創建了一個non-durable的隊列，然后想把它改變成durable的，唯一的辦法就是刪除這個隊列然后重現創建;
* RabbitMQ消息模型的核心理念是:發布者(producer)不會直接發送任何消息給隊列。
* 事實上，發布者(producer)甚至不知道消息是否已經被投遞到隊列。
* 發布者(producer)只需要把消息發送給一個exchange。
* exchange非常簡單，它一邊從發布者方接收消息，一邊把消息推入隊列。
* exchange必須知道如何處理它接收到的消息，是應該推送到指定的隊列還是是多個隊列，或者是直接忽略消息。
* 這些規則是通過exchange type來定義的;
* @author
*/
public class RabbitUtil1 {
/**日志對象**/
private final static Logger log = Logger.getLogger(RabbitUtil1.class);
/**RabbitMq連接工廠對象**/
private volatile static ConnectionFactory factory = null;
public static boolean stopRabbitFlag=false;
/**構造方法**/
public RabbitUtil1() {
this(ConnectionFactory.DEFAULT_HOST,
ConnectionFactory.DEFAULT_AMQP_PORT,
ConnectionFactory.DEFAULT_VHOST,
ConnectionFactory.DEFAULT_USER, ConnectionFactory.DEFAULT_PASS);
}
/**
* 構造方法
* @param serverHost Rabbit服務主機
*/
public RabbitUtil1(String serverHost) {
this(serverHost, ConnectionFactory.DEFAULT_AMQP_PORT,
ConnectionFactory.DEFAULT_VHOST,
ConnectionFactory.DEFAULT_USER, ConnectionFactory.DEFAULT_PASS);
}
/**
* 構造方法
* @param serverHost Rabbit服務主機
* @param username 用戶名
* @param password 密碼
*/
public RabbitUtil1(String serverHost, String username, String password) {
this(serverHost, ConnectionFactory.DEFAULT_AMQP_PORT,
ConnectionFactory.DEFAULT_VHOST, username, password);
}
/**
* 構造方法
* @param serverHost Rabbit服務主機
* @param vhost 虛擬host(類似權限組)
* @param username 用戶名
* @param password 密碼
*/
public RabbitUtil1(String serverHost, String vhost, String username,
String password) {
this(serverHost, ConnectionFactory.DEFAULT_AMQP_PORT, vhost, username,
password);
}
/**
* 構造方法
* @param serverHost Rabbit服務主機
* @param port Rabbit端口
* @param username 用戶名
* @param password 密碼
*/
public RabbitUtil1(String serverHost, int port, String username,
String password) {
this(serverHost, port, ConnectionFactory.DEFAULT_VHOST, username,
password);
}
/**
* 構造方法(初始化單例RabbitConnectionFactory)
* @param serverHost Rabbit服務主機
* @param port Rabbit端口
* @param vhost 虛擬host(類似權限組)
* @param username 用戶名
* @param password 密碼
*/
public RabbitUtil1(String serverHost, int port, String vhost,
String username, String password) {
if (null == factory) {
synchronized (ConnectionFactory.class) {
if (null == factory) {
factory = new ConnectionFactory();
factory.setHost(serverHost);
factory.setPort(port);
factory.setVirtualHost(vhost);
factory.setUsername(username);
factory.setPassword(password);
log.info(">>>>>>Singleton ConnectionFactory Create Success>>>>>>");
}
}
}
if(stopRabbitFlag){
stopRabbitFlag=false;
}
}
/**
* 創建連接
* @return 連接
* @throws Exception
*/
private Connection buildConnection() throws Exception {
return factory.newConnection();
}
/**
* 創建信道
* @param connection 連接
* @return 信道
* @throws Exception 運行時異常
*/
private Channel buildChannel(Connection connection) throws Exception {
return connection.createChannel();
}
/**
* 關閉連接和信道
* @param connection rabbitmq連接
* @param channel rabbitmq信道
*/
private void close(Connection connection, Channel channel) {
try {
if (null != channel) {
channel.close();
}
if (null != connection) {
connection.close();
}
} catch (Exception e) {
log.error(">>>>>>關閉RabbitMq的connection或channel發生異常>>>>>>", e);
}
}
/**
* 發送direct類型消息
* @param exchangeName exchange名稱
* @param routingKey 路由key字符串
* @param message 待發送的消息
* @throws Exception 運行時異常
*/
public void sendDirect(String exchangeName, String routingKey, String message) throws Exception {
Connection connection = null;
Channel channel = null;
try {
connection = buildConnection();
channel = buildChannel(connection);
channel.basicPublish(exchangeName, routingKey,
MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
log.info("消息(" + message + "發布成功");
} finally {
close(connection, channel);
}
}
/**
* 接收direct類型消息(手動發送ack)
* @param queueName 隊列名稱
* @param processer 回調處理接口
* @throws Exception 運行時異常
*/
public void receiveDirect(String queueName, RabbitCallback processer) throws Exception {
receiveDirect(queueName, false, processer);
}
/**
* 接收direct類型消息
* @param queueName 隊列名稱
* @param autoAck 是否自動發送ack true-是 false-否
* @param processer 回調處理接口
* @throws Exception 運行時異常
*/
public void receiveDirect(String queueName, boolean autoAck, RabbitCallback processer)
throws Exception {
basicConsume(queueName, autoAck, processer);
}
/**
* 循環獲取消息并處理
* @param queueName 隊列名稱
* @param autoAck 是否自動發送ack true-是 false-否
* @param processer 回調處理接口
* @throws Exception 運行時異常
*/
private void basicConsume(String queueName, final boolean autoAck, final RabbitCallback processer)
throws Exception {
final Channel channel = buildChannel(buildConnection());
/**channel.basicQos(1)
* 保證接收端僅在發送了ack之后才會接收下一個消息，
* 在這種情況下發送端會嘗試把消息發送給下一個接收端
*/
channel.basicQos(1);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
long deliveryTag = envelope.getDeliveryTag();
String responseMsg = new String(body, "UTF-8");
System.out.println("Consumer===");
if (stopRabbitFlag){
if(channel!=null){
try {
channel.basicNack(deliveryTag, false, true);//回復處理失敗，客戶需要關閉
channel.close();//關閉客戶端
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}else {//若服務停止標志為false,需要處理業務，調用回調函數處理
boolean success = processer.process(responseMsg);
if (!autoAck) {
if (success) {
channel.basicAck(deliveryTag, false);
} else {
channel.basicNack(deliveryTag, false, true);
}
}
}
}
};
channel.basicConsume(queueName, autoAck, consumer);
}
}
總結
以上是生活随笔為你收集整理的java rabbitmq 工具类_RabbitMq通用管理工具类的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： java课程设计 成绩_java课程设计
- 下一篇： java treeset 红黑树_【数据