day75_淘淘商城專案_08_同步索引庫問題分析 + ActiveMQ介紹/安裝/使用 + ActiveMQ整合spring + 使用ActiveMQ實現新增商品後同步索引庫_匠心筆記
淘淘商城專案_08
課程計劃
- 1、同步索引庫問題分析
- 2、什麼是MQ
- 3、ActiveMQ的安裝
- 4、ActiveMQ的使用方法
- 5、ActiveMQ整合spring
- 6、使用訊息佇列實現新增商品後同步索引庫
1、同步索引庫問題分析
方案一：在taotao-manager中，新增商品的業務邏輯中，新增一個同步索引庫的業務邏輯。
缺點：業務邏輯耦合度
方案二：業務邏輯在taotao-search中實現，呼叫服務在taotao-manager實現。業務邏輯分開。既可以是服務的生產者，也可以是服務的消費者。
缺點：
服務之間的耦合度變高。服務的啟動有先後順序。隨著呼叫的服務會越來越多，服務之間的呼叫越來越複雜，難以管理。方案三：使用訊息佇列。MQ是一個訊息中介軟體。
MQ就相當於一個祕書、一個運營商，如下圖所示：
存在的問題：
1、如果MQ掛了，所有相關的服務都掛了。
2、MQ有效能的瓶頸，儘量減少訊息的內容的大小。
注意：
技術的選型和具體的業務有關，只選擇合適的技術。如果MQ掛了：
1、通過日誌查詢原因。
2、通知相關的人員修復。
3、關鍵的業務必須保證有備用方案。
2、ActiveM的介紹
MQ是一個訊息中介軟體，比如：ActiveMQ(Java)、RabbitMQ(Erlang)、Kafka(大資料)都屬於MQ，是MQ的產品。
2.1、什麼是ActiveMQ
ActiveMQ 是 Apache 出品，是最流行的，能力強勁的
開源訊息匯流排。ActiveMQ 是一個完全支援JMS1.1和J2EE 1.4規範的JMS Provider 實現，儘管JMS規範出臺已經是很久的事情了，但是JMS在當今的J2EE應用中間仍然扮演著特殊的地位。
主要特點：
1. 多種語言和協議編寫客戶端。語言: Java, C, C++, C#, Ruby, Perl, Python, PHP。應用協議: OpenWire, Stomp REST, WS Notification, XMPP, AMQP。
2. 完全支援JMS1.1和J2EE 1.4規範 (持久化、XA訊息、事務)。
3. 對spring的支援，ActiveMQ可以很容易內嵌到使用Spring的系統裡面去，而且也支援Spring2.0的特性。
4. 通過了常見J2EE伺服器(如：Geronimo, JBoss 4, GlassFish, WebLogic)的測試，其中通過 JCA 1.5 resource adaptors 的配置，可以讓ActiveMQ可以自動的部署到任何相容J2EE 1.4 商業伺服器上。
5. 支援多種傳送協議:in-VM, TCP,SSL, NIO,UDP, JGroups, JXTA。
6. 支援通過 JDBC 和 journal 提供高速的訊息持久化。
7. 從設計上保證了高效能的叢集，客戶端-伺服器，點對點。
8. 支援Ajax。
9. 支援與Axis的整合。
10. 可以很容易呼叫內嵌 JMS provider，進行測試。
2.2、ActiveMQ的訊息形式
對於訊息的傳遞有兩種型別：
一種是點對點的，即一個生產者和一個消費者一一對應。（簡訊）
另一種是釋出/訂閱模式，即一個生產者產生訊息並進行傳送後，可以由多個消費者進行接收。（廣播）
JMS定義了五種不同的訊息正文格式，以及呼叫的訊息型別，允許你傳送並接收以一些不同形式的資料，提供現有訊息格式的一些級別的相容性。
StreamMessage --> Java原始值的資料流
MapMessage --> 一套名稱-值對
TextMessage --> 一個字串物件
ObjectMessage --> 一個序列化的 Java物件
BytesMessage --> 一個位元組的資料流
3、ActiveMQ的安裝
網址：http://activemq.apache.org/
下載ActiveMQ，使用的版本是5.13.0，最新的版本是5.15.8。
3.1、安裝環境
1、activemq是java語言開發的，需要安裝jdk。
2、安裝Linux系統。生產環境都是Linux系統。
3.2、安裝步驟
第一步：把 apache-activemq-5.13.0-bin.tar.gz 的壓縮包上傳到Linux系統。
第二步：解壓縮後刪除壓縮包。
[[email protected] ~]# ll
總用量 65472
-rw-r--r--. 1 root root 50385720 11月 25 12:32 apache-activemq-5.13.0-bin.tar.gz
[[email protected] ~]# tar zxf apache-activemq-5.13.0-bin.tar.gz
[[email protected] ~]# ll
總用量 52360
drwxr-xr-x. 10 root root 193 11月 30 2015 apache-activemq-5.13.0
-rw-r--r--. 1 root root 53613208 1月 7 2016 apache-activemq-5.13.0-bin.tar.gz
[[email protected] ~]# rm -rf apache-activemq-5.13.0-bin.tar.gz
第三步：啟動。
使用bin目錄下的activemq命令
啟動：
[[email protected] bin]# ./activemq start
關閉：
[[email protected] bin]# ./activemq stop
檢視狀態：
[[email protected] bin]# ./activemq status
檢視執行埠號：
[[email protected] bin]# ps aux | grep activemq
第四步：本博主的是CentOS7，防火牆使用的是firewalld，我們使用修改配置檔案的方式來新增用到的埠（修改後需要重啟firewalld服務）
[[email protected] bin]# cd /etc/firewalld/zones/
[[email protected] zones]# pwd
/etc/firewalld/zones
[[email protected] zones]# vim public.xml
編輯public.xml檔案，新增程式碼：<port protocol="tcp" port="8161"/>
儲存退出後，然後我們需要重啟firewalld服務：
[[email protected] zones]# service firewalld restart
Redirecting to /bin/systemctl restart firewalld.service
[[email protected] zones]#
第五步：進入activemq管理後臺。
訪問地址：http://192.168.25.168:8161/admin
使用者名稱：admin
密碼：admin
如下圖所示：
解決ActiveMQ訪問後臺時出現503錯誤（點選Queues按鈕時）：
問題原因：機器名沒有對映到ip地址127.0.0.1上。
1、檢視機器名
[[email protected] bin]# vim /etc/sysconfig/network
NETWORKING=yes
HOSTNAME=taotao.com
2、修改機器名和ip的對映關係，即修改hosts檔案
[[email protected] bin]# vim /etc/hosts
127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4 taotao
::1 localhost localhost.localdomain localhost6 localhost6.localdomain6
3、重啟ActiveMQ服務，問題即可解決
4、ActiveMQ的使用方法
點對點：
釋出/訂閱：
4.1、Queue（佇列）
ActiveMQ的Queue方式預設在ActiveMQ的服務端是持久化快取的。
4.1.1、Producer
本例中，我們使用taotao-manager-service工程作為生產者。
生產者：生產訊息，傳送端。
把jar包新增到工程中。使用5.11.2版本的jar包。版本號我們在taotao-parent中的pom.xml中配置。
第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號61616。注意：8161是ActiveMQ的web服務的埠。
第二步：使用ConnectionFactory物件建立一個Connection物件。
第三步：開啟連線，呼叫Connection物件的start()方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Queue物件。
第六步：使用Session物件建立一個Producer物件。
第七步：建立一個Message物件，建立一個TextMessage物件。
第八步：使用Producer物件傳送訊息。
第九步：關閉資源。
我們在taotao-manager-service工程中新建測試類和測試方法。
測試程式碼如下：
/**
* ActiveMQ的Queue方式的Producer
* @throws Exception
*/
@Test
public void queueProducerTest() throws Exception {
// 第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號61616。注意：8161是ActiveMQ的web服務的埠。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616"); // 注意：虛擬機器上的安裝ActiveMQ服務要新增該埠
// 第二步：使用ConnectionFactory物件建立一個Connection物件。
Connection connection = connectionFactory.createConnection();
// 第三步：開啟連線，呼叫Connection物件的start方法。
connection.start();
// 第四步：使用Connection物件建立一個Session物件。
// 第一個引數：是否開啟分散式事務。true：開啟分散式事務，當第一個引數為true時，第二個引數忽略。
// 一般我們不使用分散式事務，因為分散式事務複雜，效能不好。網際網路專案要求快速響應給使用者，其餘的事情交給訊息佇列向各個服務發訊息讓他們去做。這樣保證了資料的最終一致。
// 第二個引數：當第一個引數為false時，第二個引數才有意義，表示訊息的應答模式。1、自動應答 2、手動應答。一般是自動應答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Queue物件。引數表示佇列的名稱。
Queue queue = session.createQueue("test-queue");
// 第六步：使用Session物件建立一個Producer物件。
MessageProducer producer = session.createProducer(queue);
// 第七步：建立一個Message物件，建立一個TextMessage物件。
// 方式一：
// TextMessage message = new ActiveMQTextMessage();
// message.setText("hello ActiveMQ, this is my first test.");
// 方式二：
TextMessage textMessage = session.createTextMessage("hello ActiveMQ, this is my first test.");
// 第八步：使用Producer物件傳送訊息。
producer.send(textMessage);
// 第九步：關閉資源。
producer.close();
session.close();
connection.close();
}
進入activemq管理後臺檢視新增的queues佇列，如下圖：
4.1.2、Consumer
本例中，我們使用taotao-search-service工程作為消費者。
消費者：接收訊息。
第一步：建立一個ConnectionFactory物件。
第二步：從ConnectionFactory物件中獲得一個Connection物件。
第三步：開啟連線。呼叫Connection物件的start()方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件。和傳送端保持一致queue，並且佇列的名稱一致。
第六步：使用Session物件建立一個Consumer物件。
第七步：接收訊息。
第八步：列印訊息。
第九步：關閉資源。
測試程式碼如下：
/**
* ActiveMQ的Queue方式的Consumer
* @throws Exception
*/
@Test
public void queueConsumerTest() throws Exception {
// 第一步：建立一個ConnectionFactory物件。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：從ConnectionFactory物件中獲得一個Connection物件。
Connection connection = connectionFactory.createConnection();
// 第三步：開啟連線。呼叫Connection物件的start方法。
connection.start();
// 第四步：使用Connection物件建立一個Session物件。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session物件建立一個Destination物件。和傳送端保持一致queue，並且佇列的名稱一致。
Queue queue = session.createQueue("test-queue");
// 第六步：使用Session物件建立一個Consumer物件。
MessageConsumer consumer = session.createConsumer(queue);
// 第七步：接收訊息。
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
String text = null;
// 取訊息的內容
text = textMessage.getText();
// 第八步：列印訊息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
// 系統等待接收訊息：
// 方式一：
/*
while (true) {
Thread.sleep(100);
}*/
// 方式二：等待鍵盤輸入
System.in.read();
// 第九步：關閉資源
consumer.close();
session.close();
connection.close();
}
4.2、Topic（話題）
ActiveMQ的Topic方式與ActiveMQ的Queue方式大部分都是一樣的。
ActiveMQ的Topic方式預設在ActiveMQ的服務端是不持久化的。如果消費者客戶端沒有接受到訊息內容，訊息內容就會丟了。所以消費者客戶端需要先在啟動狀態（Topic方式可以有多個消費者）。我們也可以修改配置將Topic方式持久化快取。後面可以自己學習。
4.2.1、Producer
使用步驟：
第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號。
第二步：使用ConnectionFactory物件建立一個Connection物件。
第三步：開啟連線，呼叫Connection物件的start方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Topic物件。
第六步：使用Session物件建立一個Producer物件。
第七步：建立一個Message物件，建立一個TextMessage物件。
第八步：使用Producer物件傳送訊息。
第九步：關閉資源。
測試程式碼如下：
/**
* ActiveMQ的Topic方式的Consumer
* @throws Exception
*/
@Test
public void topicProducerTest() throws Exception {
// 第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號61616。注意：8161是ActiveMQ的web服務的埠。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616"); // 注意：虛擬機器上的安裝ActiveMQ服務要新增該埠
// 第二步：使用ConnectionFactory物件建立一個Connection物件。
Connection connection = connectionFactory.createConnection();
// 第三步：開啟連線，呼叫Connection物件的start方法。
connection.start();
// 第四步：使用Connection物件建立一個Session物件。
// 第一個引數：是否開啟分散式事務。true：開啟分散式事務，當第一個引數為true時，第二個引數忽略。
// 一般我們不使用分散式事務，因為分散式事務複雜，效能不好。網際網路專案要求快速響應給使用者，其餘的事情交給訊息佇列向各個服務發訊息讓他們去做。這樣保證了資料的最終一致。
// 第二個引數：當第一個引數為false時，第二個引數才有意義，表示訊息的應答模式。1、自動應答 2、手動應答。一般是自動應答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Topic物件。引數表示話題的名稱。
Topic topic = session.createTopic("test-topic");
// 第六步：使用Session物件建立一個Producer物件。
MessageProducer producer = session.createProducer(topic);
// 第七步：建立一個Message物件，建立一個TextMessage物件。
// 方式一：
// TextMessage message = new ActiveMQTextMessage();
// message.setText("hello ActiveMQ, this is my first test.");
// 方式二：
TextMessage textMessage = session.createTextMessage("hello ActiveMQ, this is my first test.");
// 第八步：使用Producer物件傳送訊息。
producer.send(textMessage);
// 第九步：關閉資源。
producer.close();
session.close();
connection.close();
}
4.2.2、Consumer
消費者：接收訊息。
第一步：建立一個ConnectionFactory物件。
第二步：從ConnectionFactory物件中獲得一個Connection物件。
第三步：開啟連線。呼叫Connection物件的start方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件。和傳送端保持一致topic，並且話題的名稱一致。
第六步：使用Session物件建立一個Consumer物件。
第七步：接收訊息。
第八步：列印訊息。
第九步：關閉資源。
測試程式碼如下：
/**
* ActiveMQ的Topic方式的Consumer
* @throws Exception
*/
@Test
public void topicConsumerTest() throws Exception {
// 第一步：建立一個ConnectionFactory物件。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：從ConnectionFactory物件中獲得一個Connection物件。
Connection connection = connectionFactory.createConnection();
// 第三步：開啟連線。呼叫Connection物件的start方法。
connection.start();
// 第四步：使用Connection物件建立一個Session物件。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session物件建立一個Destination物件。和傳送端保持一致Topic，並且佇列的名稱一致。
Topic topic = session.createTopic("test-topic");
// 第六步：使用Session物件建立一個Consumer物件。
MessageConsumer consumer = session.createConsumer(topic);
// 第七步：接收訊息。
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
String text = null;
// 取訊息的內容
text = textMessage.getText();
// 第八步：列印訊息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
System.out.println("topic的消費者客戶端01...");
// 系統等待接收訊息：
// 方式一：
/*
while (true) {
Thread.sleep(100);
}*/
// 方式二：等待鍵盤輸入
System.in.read();
// 第九步：關閉資源
consumer.close();
session.close();
connection.close();
}
注意：
Topic 預設是不存在於MQ伺服器中的，一旦傳送之後，如果沒有訂閱，就沒了。
Queue 預設是存在於MQ的伺服器中的，傳送訊息之後，隨時取。但是一定是一個消費者取完就沒了。
5、ActiveMQ整合spring
5.1、使用方法
第一步：把Activemq模板相關的jar包新增到工程中。在Maven工程中是新增依賴jar包。最開始配置spring時已經配置過了。
第二步：編寫配置檔案applicationContext-activemq.xml。
第三步：配置生產者。使用JMSTemplate物件。傳送訊息。
第四步：在spring容器中配置Destination。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.2.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-4.2.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-4.2.xsd
http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util-4.2.xsd">
<!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.25.168:61616"></property>
</bean>
<!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="targetConnectionFactory"></property>
</bean>
<!-- 接收和傳送訊息時使用的類 -->
<!-- 配置訊息的生產者 -->
<!-- Spring提供的JMS工具類，它可以進行訊息傳送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory物件 -->
<property name="connectionFactory" ref="connectionFactory"></property>
</bean>
<!-- 配置訊息的Destination物件 -->
<!-- 這個是佇列目的地，點對點的 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg name="name" value="spring-test-queue"></constructor-arg>
</bean>
<!-- 這個是話題目的地，一對多的 -->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg name="name" value="spring-test-topic"></constructor-arg>
</bean>
</beans>
第五步：程式碼測試。
5.2、程式碼測試（Queue方式）
5.2.1、傳送訊息
taotao-manager-service中傳送訊息。
第一步：初始化一個spring容器。即載入配置檔案applicationContext-activemq.xml。
第二步：從容器中獲得JMSTemplate物件。
第三步：從容器中獲得一個Destination物件。
第四步：使用JMSTemplate物件傳送訊息，需要知道Destination。
測試程式碼如下：
/**
* ActiveMQ整合spring，ActiveMQ的Queue方式的Producer
* @throws Exception
*/
@Test
public void queueProducerTest() throws Exception {
// 第一步：初始化一個spring容器，即載入配置檔案applicationContext-activemq.xml。
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath:spring/applicationContext-activemq.xml");
// 第二步：從容器中獲得JMSTemplate物件。
JmsTemplate jmsTemplate = applicationContext.getBean(JmsTemplate.class);
// 第三步：從容器中獲得一個Destination物件。
Queue queue = (Queue) applicationContext.getBean("queueDestination"); // 建議根據id取
// 第四步：使用JMSTemplate物件傳送訊息，需要知道Destination。
jmsTemplate.send(queue, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage("spring activemq test");
return textMessage;
}
});
}
5.2.2、接收訊息
taotao-search-service中接收訊息。
第一步：把jar包新增到工程中。使用5.11.2版本的jar包。版本號我們在taotao-parent中的pom.xml中配置。
第二步：建立一個MessageListener的實現類。用於接收ActiveMQ傳送的訊息。
/**
* 自定義的MessageListener的實現類，用於接收ActiveMQ傳送的訊息
* @author chenmingjun
* @date 2018年11月25日下午11:30:56
* @version 1.0
*/
public class MyMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
// 接收到訊息，進行消費
try {
TextMessage textMessage = (TextMessage) message;
// 取訊息內容
String text = textMessage.getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
第三步：編寫配置檔案applicationContext-activemq.xml，配置spring和Activemq整合，配置訊息的消費者。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.2.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-4.2.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-4.2.xsd
http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util-4.2.xsd">
<!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->