ActiveMQ使用場景？動(dòng)力節(jié)點(diǎn)小編來告訴大家。以下介紹消息隊(duì)列在實(shí)際應(yīng)用中常用的使用場景。

異步處理，應(yīng)用解耦，流量削鋒和消息通訊四個(gè)場景。本篇使用ActiveMQ+SpringBoot來模擬這四個(gè)場景。

1.異步處理

場景說明：汽車觸發(fā)圍欄報(bào)警后，需要發(fā)送報(bào)警郵件和報(bào)警短信。傳統(tǒng)的做法有兩種1.串行的方式;2.并行方式。

(1)串行方式：將報(bào)警信息寫入數(shù)據(jù)庫成功后，發(fā)送報(bào)警郵件，再發(fā)送報(bào)警短信。以上三個(gè)任務(wù)全部完成后，該報(bào)警信息加入統(tǒng)計(jì)列表。

(2)并行方式：報(bào)警信息寫入數(shù)據(jù)庫成功后，同時(shí)發(fā)送報(bào)警郵件和短信。

假設(shè)三個(gè)業(yè)務(wù)節(jié)點(diǎn)每個(gè)使用50毫秒鐘，不考慮網(wǎng)絡(luò)等其他開銷，則串行方式的時(shí)間是150毫秒，并行的時(shí)間可能是100毫秒。

因?yàn)镃PU在單位時(shí)間內(nèi)處理的請求數(shù)是一定的，假設(shè)CPU1秒內(nèi)吞吐量是100次。則串行方式1秒內(nèi)CPU可處理的請求量是7次(1000/150)。并行方式處理的請求量是10次(1000/100)。

小結(jié)：如以上案例描述，傳統(tǒng)的方式系統(tǒng)的性能(并發(fā)量，吞吐量，響應(yīng)時(shí)間)會(huì)有瓶頸。如何解決這個(gè)問題呢?

引入消息隊(duì)列，將不是必須的業(yè)務(wù)邏輯，異步處理。改造后的架構(gòu)如下：

代碼示例

(1)在pom文件中引入activemq依賴

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
        <version>1.5.6.RELEASE</version>
    </dependency>

(2)在配置文件中加上activemq的配置

spring.activemq.broker-url=tcp://127.0.0.1:61616
# 在考慮結(jié)束之前等待的時(shí)間
#spring.activemq.close-timeout=15s 
# 默認(rèn)代理URL是否應(yīng)該在內(nèi)存中。如果指定了顯式代理，則忽略此值。
spring.activemq.in-memory=true 
# 是否在回滾回滾消息之前停止消息傳遞。這意味著當(dāng)啟用此命令時(shí)，消息順序不會(huì)被保留。
spring.activemq.non-blocking-redelivery=false
# 密碼
spring.activemq.password=123456
# 等待消息發(fā)送響應(yīng)的時(shí)間。設(shè)置為0等待永遠(yuǎn)。
spring.activemq.send-timeout=0
spring.activemq.user=haha
# 是否信任所有包
#spring.activemq.packages.trust-all=
# 要信任的特定包的逗號分隔列表（當(dāng)不信任所有包時(shí)）
#spring.activemq.packages.trusted=
# 當(dāng)連接請求和池滿時(shí)是否阻塞。設(shè)置false會(huì)拋“JMSException異常”。
#spring.activemq.pool.block-if-full=true
# 如果池仍然滿，則在拋出異常前阻塞時(shí)間。
#spring.activemq.pool.block-if-full-timeout=-1ms
# 是否在啟動(dòng)時(shí)創(chuàng)建連接。可以在啟動(dòng)時(shí)用于加熱池。
#spring.activemq.pool.create-connection-on-startup=true
# 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
#spring.activemq.pool.enabled=false 
# 連接過期超時(shí)。
#spring.activemq.pool.expiry-timeout=0ms
# 連接空閑超時(shí)
#spring.activemq.pool.idle-timeout=30s
# 連接池最大連接數(shù)
#spring.activemq.pool.max-connections=1
# 每個(gè)連接的有效會(huì)話的最大數(shù)目。
#spring.activemq.pool.maximum-active-session-per-connection=500
# 當(dāng)有"JMSException"時(shí)嘗試重新連接
#spring.activemq.pool.reconnect-on-exception=true
# 在空閑連接清除線程之間運(yùn)行的時(shí)間。當(dāng)為負(fù)數(shù)時(shí)，沒有空閑連接驅(qū)逐線程運(yùn)行。
#spring.activemq.pool.time-between-expiration-check=-1ms
# 是否只使用一個(gè)MessageProducer
#spring.activemq.pool.use-anonymous-producers=true

(3)消息生產(chǎn)者

import java.util.Map;
import javax.jms.Destination;
import javax.jms.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessagePostProcessor;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
 * 報(bào)警消息Producer
 * @author ko
 *
 */
@Component
//@EnableScheduling
public class AlarmProducer {    
    // 也可以注入JmsTemplate，JmsMessagingTemplate對JmsTemplate進(jìn)行了封裝
    @Autowired
    private JmsTemplate jmsTemplate;
//    private JmsMessagingTemplate jmsTemplate;    
//    @Autowired
//    private Queue queue;    
//    @Scheduled(fixedDelay=5000) // 5s執(zhí)行一次   只有無參的方法才能用該注解
    public void sendMessage(Destination destination, String message){
//        jmsTemplate.convertAndSend(destinationName, payload, messagePostProcessor);
        this.jmsTemplate.convertAndSend(destination, message);
    }
　　// 雙向隊(duì)列
　　　　// @JmsListener(destination="out.queue") 
　　　　// 　　public void consumerMessage(String text){  
    　　　　// 　　System.out.println("從out.queue隊(duì)列收到的回復(fù)報(bào)文為:"+text);  
　　　　// }
}

(4)controller里寫上測試接口

@Autowired
    private AlarmProducer alarmProducer;    
    @RequestMapping(value="/chufabaojing",method=RequestMethod.GET)
    @ApiOperation(value="觸發(fā)報(bào)警", notes="觸發(fā)報(bào)警")
    @ApiImplicitParams({
        @ApiImplicitParam(name = "devicename", value = "name",example = "xxxx", required = true, dataType = "string",paramType="query"),
    })
    public String chufabaojing(String devicename){        
        List<String> alarmStrList = new ArrayList<>();
        alarmStrList.add(devicename+"out fence01");
        alarmStrList.add(devicename+"out fence02");
        alarmStrList.add(devicename+"in fence01");
        alarmStrList.add(devicename+"in fence02");        
        System.out.println("設(shè)備"+devicename+"出圍欄報(bào)警");
        // 報(bào)警信息寫入數(shù)據(jù)庫
        System.out.println("報(bào)警數(shù)據(jù)寫入數(shù)據(jù)庫。。。");        
        // 寫入消息隊(duì)列
        Destination destination = new ActiveMQQueue("mytest.queue");
        for (String alarmStr : alarmStrList) {
            alarmProducer.sendMessage(destination, alarmStr);
        }        
        // 消息寫進(jìn)消息隊(duì)列里就不管了        
        // 下面兩步驟移到activemq消費(fèi)者里
        // 發(fā)送郵件
        // 發(fā)送短信        
        return "success";
    }

2.應(yīng)用解耦

場景介紹，在spring cloud分布式微服務(wù)項(xiàng)目中，工單管理和設(shè)備管理分別是兩個(gè)微服務(wù)，如果A工單被張三接單了，那么工單狀態(tài)要設(shè)為已派單，檢驗(yàn)員設(shè)為張三，設(shè)備狀態(tài)要置為在檢。

傳統(tǒng)的做法是，先調(diào)用工單管理的工單更新接口，成功之后再調(diào)用設(shè)備管理的設(shè)備更新接口，成功之后再返回操作提示給用戶。這樣做的缺點(diǎn)是應(yīng)用耦合，如果在派單操作的時(shí)候正好設(shè)備管理微服務(wù)掛了或者阻塞了，那么派單操作就會(huì)失敗或者要等待很長時(shí)間無反饋。另外如果設(shè)備管理的接口有變動(dòng)，那么工單管理里面的代碼也要改動(dòng)。

引入消息中間件，派單的時(shí)候，工單管理的工單更新接口處理好后把信息寫入消息隊(duì)列，然后直接返回操作反饋給用戶。不管工單管理服務(wù)正不正常，正常就從消息隊(duì)列里訂閱消息處理，不正常就等待回復(fù)正常后再訂閱消息處理。

3.流量削峰

場景介紹，XX公司的系統(tǒng)原來是針對A地區(qū)的客戶開發(fā)的，現(xiàn)在為了搶占市場，拿下了B和C兩個(gè)地區(qū)的客戶，那么新系統(tǒng)上線，就存在如何把B和C的基礎(chǔ)數(shù)據(jù)導(dǎo)入XX公司的系統(tǒng)中來的問題，短時(shí)間內(nèi)要把龐大的舊數(shù)據(jù)改造適合新系統(tǒng)再導(dǎo)入進(jìn)來，這很容易使系統(tǒng)掛掉，另外每天還有增量數(shù)據(jù)產(chǎn)生。

這時(shí)可以引入消息中間件，B和C的客戶只要負(fù)責(zé)把數(shù)據(jù)規(guī)則放到消息隊(duì)列里就好了，XX公司可以有條不紊的從消息隊(duì)列里訂閱數(shù)據(jù)，可以有效緩解短時(shí)間內(nèi)的高流量壓力，但是這也對消息中間件的可靠性提出了要求。

4.點(diǎn)對點(diǎn)通訊

類似聊天室的功能。