Springboot + ActiveMq 簡單實現
linux下搭建activeMq 參考
1 引入依賴包
pom.xml 新增以下內容
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
2 增加配置
在application.properties檔案中新增以下內容
spring.activemq.broker-url=tcp://10.8.206.178:61616 spring.activemq.user=admin spring.activemq.password=admin #無超時時間 spring.activemq.pool.enabled=false #(為true時是topic模式，為false是queue模式) jms.pub-sub-domain=false
當然還有許多引數可以新增 ，本人並沒有全部測試
# 在考慮結束之前等待的時間 #spring.activemq.close-timeout=15s # 預設代理URL是否應該在記憶體中。如果指定了顯式代理，則忽略此值。 spring.activemq.in-memory=true # 是否在回滾回滾訊息之前停止訊息傳遞。這意味著當啟用此命令時，訊息順序不會被保留。 spring.activemq.non-blocking-redelivery=false # 密碼 spring.activemq.password=123456 # 等待訊息傳送響應的時間。設定為0等待永遠。 spring.activemq.send-timeout=0 spring.activemq.user=haha # 是否信任所有包 #spring.activemq.packages.trust-all= # 要信任的特定包的逗號分隔列表（當不信任所有包時） #spring.activemq.packages.trusted= # 當連線請求和池滿時是否阻塞。設定false會拋“JMSException異常”。 #spring.activemq.pool.block-if-full=true # 如果池仍然滿，則在丟擲異常前阻塞時間。 #spring.activemq.pool.block-if-full-timeout=-1ms # 是否在啟動時建立連線。可以在啟動時用於加熱池。 #spring.activemq.pool.create-connection-on-startup=true # 是否用Pooledconnectionfactory代替普通的ConnectionFactory。 #spring.activemq.pool.enabled=false # 連線過期超時。 #spring.activemq.pool.expiry-timeout=0ms # 連線空閒超時 #spring.activemq.pool.idle-timeout=30s # 連線池最大連線數 #spring.activemq.pool.max-connections=1 # 每個連線的有效會話的最大數目。 #spring.activemq.pool.maximum-active-session-per-connection=500 # 當有"JMSException"時嘗試重新連線 #spring.activemq.pool.reconnect-on-exception=true # 在空閒連線清除執行緒之間執行的時間。當為負數時，沒有空閒連線驅逐執行緒執行。 #spring.activemq.pool.time-between-expiration-check=-1ms # 是否只使用一個MessageProducer #spring.activemq.pool.use-anonymous-producers=true
3 搭建生產者
- controller類
import javax.jms.Destination; import org.apache.activemq.command.ActiveMQQueue; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.RestController; @RestController public class controller { @Autowired MqService mqservice; @RequestMapping(value = "/sendMsg",method = RequestMethod.GET) @ResponseBody public String sendMsg(){ Destination destination = new ActiveMQQueue("test.queue");; mqservice.sendMessage("Mq你好", destination); return "Mq訊息傳送成功"; } }
- Service類
import javax.jms.Destination;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;
@Service
public class MqService {
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;
void sendMessage(String msg,Destination destination){
this.jmsMessagingTemplate.convertAndSend(destination, msg);
}
}
測試：已經生產了一個訊息在佇列中
搭建消費者
- 建立監聽消費，即有訊息在佇列中會自動消費
service類
import javax.jms.Destination;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;
@Service
public class MqServer {
private JmsMessagingTemplate jms;
@JmsListener(destination = "test.queue")
public void receiveQueue(String text) {
System.err.println("收到訊息"+text);
// return text;
}
}
測試
登陸控制檯檢視
- 使用主動消費策略，即客戶自己主動消費佇列中的訊息
我們把程式碼寫在contrller類中
import javax.jms.Destination;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;
@RestController
public class controller {
@Autowired
MqServer mqserver;
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;
@RequestMapping(value = "/getMsg",method = RequestMethod.GET)
@ResponseBody
public String getMsg(){
Message string=this.jmsMessagingTemplate.receive("test.queue1");
return "Mq訊息接受成功 "+string.getPayload() ;
}
}
測試
傳送http://localhost:8989/getMsg
登陸控制檯
注意事項：
注意我們在消費佇列和生產佇列中選擇的策略（點對點/訂閱）
思考的問題：
- 1 佇列的長度是怎麼設定的，超出佇列長度後訊息會丟失嗎？
他寫的真好
佇列大大小由jvm記憶體控制，而jvm記憶體在/bin/linux/wrapper.conf檔案中設定
JVM的初始記憶體和最大記憶體
在/conf/activemq.xml檔案中
<systemUsage>
<memoryUsage>
<memoryUsage percentOfJvmHeap="70" />
</memoryUsage>
<storeUsage>
<storeUsage limit="100 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="50 gb"/>
</tempUsage>
</systemUsage>
-
systemUsage
sendFailIfNoSpaceAfterTimeout：當ActiveMQ收到一條訊息時，如果ActiveMQ這時已經沒有多餘“容量”了，那麼就會等待一段時間（這裡設定的毫秒數），如果超過這個等待時間ActiveMQ仍然沒有可用的容量，那麼就拒絕接收這條訊息並在訊息的傳送端丟擲javax.jms.ResourceAllocationException異常；
sendFailIfNoSpace，當ActiveMQ收到一條訊息時，如果ActiveMQ這時已經沒有多餘“容量”了，就直接拒絕這條訊息（不用等待一段時間），並在訊息的傳送端丟擲javax.jms.ResourceAllocationException異常。 -
memoryUsage
該子標記設定整個ActiveMQ節點的“可用記憶體限制”。這個值不能超過上文中您設定的JVM maxmemory的值。其中的percentOfJvmHeap屬性表示使用“百分數值”進行設定，除了這個屬性以外，您還可以使用limit屬性進行固定容量授權，例如：limit=”1000 mb”。這些記憶體容量將供所有佇列使用。 -
storeUsage：
該標記設定整個ActiveMQ節點，用於儲存“持久化訊息”的“可用磁碟空間”。該子標記的limit屬性必須要進行設定。在使用後續介紹的KahaDB方案或者LevelDB方案進行PERSISTENT Message持久化儲存時，這個storeUsage屬性都會起作用；但是如果使用資料庫儲存方案，這個屬性就不會起作用了。 -
tempUsage：在ActiveMQ 5.X+ 版本中，一旦ActiveMQ服務節點儲存的訊息達到了memoryUsage的限制，NON_PERSISTENT Message就會被轉儲到 temp store區域。雖然我們說過NON_PERSISTENT Message不進行持久化儲存，但是ActiveMQ為了防止“資料洪峰”出現時NON_PERSISTENT Message大量堆積致使記憶體耗盡的情況出現，還是會將NON_PERSISTENT Message寫入到磁碟的臨時區域——temp store。這個子標記就是為了設定這個temp store區域的“可用磁碟空間限制”。最後提醒各位讀者storeUsage和tempUsage並不是“最大可用空間”，而是一個閥值。
-
2 activeMq的持久化策略
預設使用kahadb ，檔案持久化
儲存位置apache-activemq-5.15.7/data/kahadb
在/conf/activemq.xml檔案可以設定持久化策略，journalMaxFileLength為檔案大小，預設為32mb
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}">
......
<persistenceAdapter>
<kahaDB directory="${activemq.data}/kahadb" journalMaxFileLength="32mb"/>
</persistenceAdapter>
......
</broker>
還可以採用資料庫持久化策略
點這裡看下
-
2 採用訂閱者模式訊息是如何消費的？
對於點對點的訊息一旦消費者完成消費這條訊息將從broker上刪除；對於釋出訂閱型別的訊息，即使所有的訂閱者都完成了消費，Broker也不一定會馬上刪除無用訊息，而是保留推送歷史，之後會非同步清除無用訊息。而每個訂閱者消費到了哪條訊息的offset會記錄在Broker，以免下次重複消費。因為訊息是順序消費，先進先出，所以只需要記錄上次訊息消費到哪裡就可以了。 -
3 activeMq、rabbitMq、kafka三者的優勢和劣勢，我們如何選擇？
點這裡
ActiveMQ: 歷史悠久的開源專案，已經在很多產品中得到應用，實現了JMS1.1規範，可以和spring-jms輕鬆融合，實現了多種協議，不夠輕巧（原始碼比RocketMQ多），支援持久化到資料庫，對佇列數較多的情況支援不好。
RabbitMq：
它比kafka成熟，支援AMQP事務處理，在可靠性上，RabbitMq超過kafka，在效能方面超過ActiveMQ。
Kafka：
Kafka設計的初衷就是處理日誌的，不支援AMQP事務處理，可以看做是一個日誌系統，針對性很強，所以它並沒有具備一個成熟MQ應該具備的特性
Kafka的效能（吞吐量、tps）比RabbitMq要強，如果用來做大資料量的快速處理是比RabbitMq有優勢的。
本人初學，有問題請指正，共同進步！