ActiveMQ發訊息、收訊息、持久化，查詢佇列剩餘訊息數、出隊數的實現

1.首先啟動activeMQ的服務

public class RunServer {
/** 啟動activeMQ服務 */
public static void main(String[] args) throws Exception {
RunServer rs = new

RunServer();
BrokerService broker = rs.startServer();
}
public

BrokerService startServer() throws Exception{
// java程式碼呼叫activemq相關的類來構造並啟動brokerService
BrokerService broker = new BrokerService();
// 以下是持久化的配置
// 持久化檔案儲存位置
File dataFilterDir = new File("targer/amq-in-action/kahadb");
KahaDBStore kaha = new KahaDBStore();
kaha.setDirectory(dataFilterDir);
// use a bigger journal file
kaha.setJournalMaxFileLength(1024*100);
// small batch means more frequent and smaller writes
kaha.setIndexWriteBatchSize(100);
// do the index write in a separate thread
kaha.setEnableIndexWriteAsync(true);
broker.setPersistenceAdapter(kaha);
// create a transport connector
broker.addConnector("tcp://localhost:61616");
broker.setUseJmx(true);
//broker.setDataDirectory("data/");
// 以下是ManagementContext的配置，從這個容器中可以取得訊息佇列中未執行的訊息數、消費者數、出隊數等等
// 設定ManagementContext
ManagementContext context = broker.getManagementContext();
context.setConnectorPort(2011);
context.setJmxDomainName("my-broker");
context.setConnectorPath("/jmxrmi");
broker.start();
System.in.read();
return broker;
}

2.傳送訊息

public class Sender {
private static final int SEND_NUMBER = 1;
public static void main(String[] args) {
// ConnectionFactory ：連線工廠，JMS 用它建立連線
ConnectionFactory connectionFactory;
// Connection ：JMS 客戶端到JMS Provider 的連線
Connection connection = null;
// Session：一個傳送或接收訊息的執行緒
Session session;
// Destination ：訊息的目的地;訊息傳送給誰.
Destination destination;
// MessageProducer：訊息傳送者
MessageProducer producer;
// TextMessage message;
// 構造ConnectionFactory例項物件，此處採用ActiveMq的實現jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://localhost:61616");
try {
// 構造從工廠得到連線物件
connection = connectionFactory.createConnection();
// 啟動
connection.start();
// 獲取操作連線
session = connection.createSession(Boolean.TRUE,
Session.AUTO_ACKNOWLEDGE);
// 獲取session注意引數值xingbo.xu-queue是一個伺服器的queue，須在在ActiveMq的console配置
destination = session.createQueue("test-persistence");
// 得到訊息生成者【傳送者】
producer = session.createProducer(destination);
// 設定不持久化，可以更改
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
// 構造訊息
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
public static void sendMessage(Session session, MessageProducer producer)
throws Exception {
for (int i = 1; i <= SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage("ActiveMq 傳送的訊息" + i);
// 傳送訊息到目的地方
System.out.println("傳送訊息:" + i);
producer.send(message);
}
}

3.收訊息

public class Receiver {
public static void main(String[] args) {
// ConnectionFactory ：連線工廠，JMS 用它建立連線
ConnectionFactory connectionFactory;
// Connection ：JMS 客戶端到JMS Provider 的連線
Connection connection = null;
// Session：一個傳送或接收訊息的執行緒
Session session;
// Destination ：訊息的目的地;訊息傳送給誰.
Destination destination;
// 消費者，訊息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
try {
// 構造從工廠得到連線物件
connection = connectionFactory.createConnection();
// 啟動
connection.start();
// 獲取操作連線
session = connection.createSession(Boolean.TRUE,
Session.AUTO_ACKNOWLEDGE);
//test-queue跟sender的保持一致，一個建立一個來接收
destination = session.createQueue("test-persistence");
consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message arg0) {
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
System.out.println("==================");
try {
System.out.println("RECEIVE1第一個獲得者:"
+ ((TextMessage) arg0).getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
} catch (Exception e) {
e.printStackTrace();
}
finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
相關推薦

ActiveMQ發訊息、收訊息、持久化，查詢佇列剩餘訊息數、出隊數的實現

1.首先啟動activeMQ的服務 public class RunServer {           /** 啟動activeMQ服務&

比較全面的Eclipse配置詳解（包括智慧提示設定、智慧提示外掛修改，修改空格自動上屏、JDK配置、各種快捷鍵列表……） - decarl - 部落格園

Eclipse編輯器基本設定 1、新增行號在邊緣處右鍵 2、改字型字型的一般配置 3、去掉拼寫錯誤檢查 4、Java程式碼風格程式碼格式化 Ctrl + Shift + F 之後點選右邊的New按鈕，新建一個風格。

比較全面的Eclipse配置詳解（包括智慧提示設定、智慧提示外掛修改，修改空格自動上屏、JDK配置、各種快捷鍵列表……）

Eclipse編輯器基本設定 1、新增行號在邊緣處右鍵 2、改字型字型的一般配置 3、去掉拼寫錯誤檢查 4、Java程式碼風格程式碼格式化 Ctrl + Shift + F 之後點選右邊的Ne

根據物料號和收貨物料憑證，查詢對應的序列號

SELECT   b~sernr c~lbbsa            FROM ser03 AS a INNER JOIN objk AS b               ON a~obknr = b~obknr inner join equi AS d on     d~sernr = b~sernr

為什麼使用訊息佇列？訊息佇列有什麼優點和缺點？Kafka、ActiveMQ、RabbitMQ、RocketMQ 都有什麼優點和缺點？

面試題為什麼使用訊息佇列？訊息佇列有什麼優點和缺點？ Kafka、ActiveMQ、RabbitMQ、RocketMQ 都有什麼區別，以及適合哪些場景？面試官心理分析其實面試官主要是想看看：第一，你知不知道你們系統裡為什麼要用訊息佇列這個東西？不少候選人，說自己專案裡用了 Redis、M

關於阿里訊息佇列RocketMQ（安裝、使用和坑），你需要知道的事情

為什麼選擇RocketMQ Apache RocketMQ作為阿里開源的一款高效能、高吞吐量的分散式訊息中介軟體。因為阿里有海量的資料量，無數業務場景的應用，是RocketMQ搶盡風頭風頭，成為不可多得中介軟體專案，加上已經正式加入Apach俱樂部，作為頂級的開源專案！一、關於

[藍芽] 6、基於nRF51822的藍芽心率計工程訊息流Log分析（詳細）

轉載自： http://www.cnblogs.com/zjutlitao/p/5051166.html 開機初始化Log Log編號

原 Win32 SDK基礎（11）—— 訊息佇列和GetMessage/PeekMessage、SendMessage/Postmesage

版權宣告：本文為博主原創文章，若轉載請註明出處；若有錯誤，歡迎指出；若有問題，歡迎留言。 https://blog.csdn.net/lzhui1987/article/details/70144952 一、訊息佇列 1.1 訊息佇列

訊息佇列MQ選型 - Kafka、RabbitMQ對比

image.png 適應場景非同步處理，應用解耦，流量削鋒和訊息通訊對比 feature scenario Kafka RabbitMQ 備註 PUB-SUB 釋出訂閱模型

C#應用WindowsApi實現查詢\列舉(FindWindow、EnumChildWindows)窗體控制元件，併發送訊息。

轉載：http://www.cnblogs.com/hehexiaoxia/p/4223927.html 首先介紹基本WindowsApi： public static extern IntPtr FindWindow(string lpClassName, strin

JMS訊息持久化，將ActiveMQ訊息持久化到mySql資料庫中

ActiveMQ5.8.0版本採用kahadb作為預設的訊息持久化方式。使用預設的持久化機制，我們不容易直接看到訊息究竟是如何持久的。ActiveMQ提供的JDBC持久化機制，能夠將持久化資訊儲存到資

Linux c 基於記憶體的程序通訊—共享記憶體、共享佇列（訊息佇列）

基於記憶體的程序通訊： 1.      核心共享記憶體程式設計模型： 1.1.建立共享記憶體，得到一個ID shmget 1.2.把ID影射成虛擬地址(掛載) shmat        1.3.使用虛擬地址訪問核心共享記憶體使用任何記憶體函式與運算子號

Spring+ActiveMQ訊息持久化，Topic持久化訂閱

訊息持久化就是將訊息儲存到磁碟，這樣的好處就是如果服務掛了，則訊息還儲存在磁碟不會丟失，服務起來後還能找到訊息並在此傳送，訊息的持久化和訊息的傳送模型是沒有關係的。訊息持久化的配置很方便的，所以其他的那些就不寫出來了，可以看看上一篇文章中的同步非同步實現方式。

Win32 SDK基礎（11）—— 訊息佇列和GetMessage/PeekMessage、SendMessage/Postmesage

一、訊息佇列 1.1 訊息佇列訊息佇列是用來存放訊息的一個佇列，訊息在佇列中先入先出，所有的視窗程式都具有訊息佇列，程式可以從佇列中獲取訊息。 1.2 訊息佇列的型別系統訊息佇列：由作業系統維護的訊息佇列，存放系統產生的訊息，如滑鼠、鍵盤

位元組跳動面試官這樣問訊息佇列：高可用、不重複消費、可靠傳輸、順序消費、訊息堆積，我整理了下

## 寫在前面又到了年底跳槽高峰季，很多小夥伴出去面試時，不少面試官都會問到訊息佇列的問題，不少小夥伴回答的不是很完美，有些小夥伴是心裡知道答案，嘴上卻沒有很好的表達出來，究其根本原因，還是對相關的知識點理解的不夠透徹。今天，我們就一起來探討下這個話題。注：文章有點長，你說你能一鼓作氣看完，我有點不信！！

解決持久化數據太大，單個節點的硬盤無法存儲的問題；解決運算量太大，單個節點的內存、CPU無法處理的問題

pro 一致性哈希普通 .html 價格 str oca 計劃硬件需要學習的技術很多，要自學新知識也不是一件容易的事，選擇一個自己比較感興趣的會是一個比較好的開端，於是，打算學一學分布式系統。　　帶著問題，有目的的學習，先了解整體架構，在深入感興趣的細節，這是我的

IIS連接數、IIS並發連接數、IIS最大並發工作線程數、應用程序池的隊列長度

這就是規範性初級展示約會第一次數量企業通用關於並發你真的了解嗎？（一）前言：對於很多工作時間短或者編程經驗不足的程序員來說，大多數會覺得並發這個詞離自己太遙遠，之所以知道並發也不過是因為受那些技術大佬成天討論並發等問題耳濡目染罷了。更有甚者，一些

QPS相關的概念收集（吞吐量（TPS）、QPS、並發數、響應時間（RT））

臺電腦接受邏輯 .cn 客戶 lan 頁面增長 value 一、概念： 1、響應時間（RT）響應時間是指系統對請求作出響應的時間。直觀上看，這個指標與人對軟件性能的主觀感受是非常一致的，因為它完整地記錄了整個計算機系統處理請求的時間。由於一個系統通常會提供許多

Nginx並發數、每秒連接數、下載速度限制，防攻擊殺手鐧

意思 bsp p s address ket agen 將在 hit sun 1.限制IP訪問頻率：HttpLimitZoneModule 限制並發連接數實例limit_zone只能定義在http作用域，limit_conn可以定義在http server location

ActiveMQ 發送和就收消息

ace listen OS factor row conn ack 多個 tar 一、添加 jar 包 <dependency> <groupId>org.apache.activemq</groupId> <ar

ActiveMQ發訊息、收訊息、持久化，查詢佇列剩餘訊息數、出隊數的實現

相關推薦