當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

ActiveMQ的介绍及使用实例.

發布時間：2024/8/26 编程问答 36 豆豆

生活随笔收集整理的這篇文章主要介紹了 ActiveMQ的介绍及使用实例. 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

今天就來說下這個項目中使用ActiveMQ的情況, MQ: message queue, 顧名思義就是消息隊列的意思.?

一: 使用場景:?

消息隊列在大型電子商務類網站，如京東、淘寶、去哪兒等網站有這深入的應用，隊列的主要作用是消除高并發訪問高峰，加快網站的響應速度。在不使用消息隊列的情況下，用戶的請求數據直接寫入數據庫，在高并發的情況下，會對數據庫造成巨大的壓力，同時也使得系統響應延遲加劇。在使用隊列后，用戶的請求發給隊列后立即返回（當然不能直接給用戶提示訂單提交成功，京東上提示：您“您提交了訂單，請等待系統確認”），再由消息隊列的消費者進程從消息隊列中獲取數據，異步寫入數據庫。由于消息隊列的服務處理速度遠快于數據庫，因此用戶的響應延遲可得到有效改善。

那么在babasport這個項目中, 我們可以在上架的時候使用消息隊列的模式:
我們之前在點擊一款商品上架的時候, 我們需要分成2步, 第一: 更新商品表中該商品的上架狀態. 第二: 將該商品信息保存到Solr服務器中. ?那么如果我們使用了消息隊列后, 第二步就可以使用發送message來異步完成.

消息隊列可以接收消息和?發送消息

消息隊列類型：

隊列：一對一聊天 ?私聊 ?QQ

主題(訂閱模式)：一對多聊天 ?群聊 ?QQ

名詞解釋:?

?二, 代碼原型
ActiveMQ需要部署到Linux系統下, 這里就不再做概述.
這里也是tar包, 導入到linux下直接解壓啟動即可, 前面已經有過很多博文講Linux下一些常用軟件的安裝步驟.

上架代碼原型:
項目構件圖:

未使用ActiveMQ前ProductServiceImpl.cs:

1 //上架 2 public void isShow(Long[] ids){ 3 Product product = new Product(); 4 product.setIsShow(true); 5 for (final Long id : ids) { 6 //上下架狀態 7 product.setId(id); 8 productDao.updateByPrimaryKeySelective(product); 9 10 //這個地方的代碼應該在babasport-solr中寫, 現在使用ActiveMQ進行遷移. 11 //TODO 保存商品信息到Solr服務器 12 SolrInputDocument doc = new SolrInputDocument(); 13 //ID 14 doc.setField("id", id); 15 //名稱 16 Product p = productDao.selectByPrimaryKey(id); 17 doc.setField("name_ik", p.getName()); 18 //圖片URL 19 doc.setField("url", p.getImgUrls()[0]); 20 //品牌 ID 21 doc.setField("brandId", p.getBrandId()); 22 //價格 sql查詢語句: select price from bbs_sku where product_id = ? order by price asc limit 1 23 SkuQuery skuQuery = new SkuQuery(); 24 skuQuery.createCriteria().andProductIdEqualTo(id); 25 skuQuery.setOrderByClause("price asc"); 26 skuQuery.setPageNo(1); 27 skuQuery.setPageSize(1); 28 List<Sku> skus = skuDao.selectByExample(skuQuery); 29 doc.setField("price", skus.get(0).getPrice()); 30 //...時間等剩下的省略 31 32 try { 33 solrServer.add(doc); 34 solrServer.commit(); 35 } catch (Exception e) { 36 // TODO Auto-generated catch block 37 e.printStackTrace(); 38 } 39 40 41 42 43 //TODO 靜態化 44 } 45 }

上面的代碼除了更改本來就該更改的商品狀態信息外, 還去見商品信息保存到了Solr服務器中了. 這里我們使用ActiveMQ進行改造:?
使用ActiveMQ后的ProductServiceImpl.cs:

1 //上架 2 public void isShow(Long[] ids){ 3 Product product = new Product(); 4 product.setIsShow(true); 5 for (final Long id : ids) { 6 //上下架狀態 7 product.setId(id); 8 productDao.updateByPrimaryKeySelective(product); 9 10 //發送商品ID到ActiveMQ即可. 11 jmsTemplate.send(new MessageCreator() { 12 13 @Override 14 public Message createMessage(Session session) throws JMSException { 15 16 return session.createTextMessage(String.valueOf(id)); 17 } 18 }); 19 20 //TODO 靜態化 21 } 22 }

接著就是配置消息發送方(JMS生產者) mq.xml:

1 <beans xmlns="http://www.springframework.org/schema/beans" 2 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:mvc="http://www.springframework.org/schema/mvc" 3 xmlns:context="http://www.springframework.org/schema/context" 4 xmlns:aop="http://www.springframework.org/schema/aop" 5 xmlns:tx="http://www.springframework.org/schema/tx" 6 xmlns:task="http://www.springframework.org/schema/task" 7 xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" 8 xsi:schemaLocation="http://www.springframework.org/schema/beans 9 http://www.springframework.org/schema/beans/spring-beans-4.0.xsd 10 http://www.springframework.org/schema/mvc 11 http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd 12 http://www.springframework.org/schema/context 13 http://www.springframework.org/schema/context/spring-context-4.0.xsd 14 http://www.springframework.org/schema/aop 15 http://www.springframework.org/schema/aop/spring-aop-4.0.xsd 16 http://www.springframework.org/schema/tx 17 http://www.springframework.org/schema/tx/spring-tx-4.0.xsd 18 http://www.springframework.org/schema/task 19 http://www.springframework.org/schema/task/spring-task-4.0.xsd 20 http://code.alibabatech.com/schema/dubbo 21 http://code.alibabatech.com/schema/dubbo/dubbo.xsd"> 22 23 24  25  26 <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 27  30 <property name="brokerURL" value="tcp://192.168.200.128:61616"/> 31  32 <property name="userName" value="admin"></property> 33 <property name="password" value="admin"></property> 34 </bean> 35 36  37 <bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean"> 38  39 <property name="connectionFactory" ref="activeMQConnectionFactory"></property> 40  41 <property name="maxConnections" value="5"></property> 42 </bean> 43 44  45 <bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> 46  47 <property name="targetConnectionFactory" ref="pooledConnectionFactoryBean"></property> 48 </bean> 49 50  51 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 52  53 <property name="connectionFactory" ref="singleConnectionFactory"></property> 54  55 <property name="defaultDestinationName" value="pId"/> 56 </bean> 57 </beans>

配置說明: 這里是首先構建一個MQ的連接工廠, 只要ActiveMQ啟動后就可以這樣構建連接了. 配置登錄的用戶名和和密碼.
接著就是配置連接池, 把連接工廠交給連接池去管理. 這些都是Apache廠商提供的.?
接著就是再將連接池交由Spring管理.?
最后我們再來配置一個jmsTemplate模板來操作ActiveMQ, 這個類似于jdbcTemplate模板. 而且我們這個里面注入了一個默認的管道, 也就是productId, 因為我們現在是傳遞消息一一去對應, 關于怎么對應 ?就是依賴于這個管道.

接下來我們就看下消息的接收方(JMS消費者)的一些東西:
消費者的目錄結構:(Solr)

Solr項目中的ActiveMQ配置文件mq.xml:

1 <beans xmlns="http://www.springframework.org/schema/beans" 2 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:mvc="http://www.springframework.org/schema/mvc" 3 xmlns:context="http://www.springframework.org/schema/context" 4 xmlns:aop="http://www.springframework.org/schema/aop" 5 xmlns:tx="http://www.springframework.org/schema/tx" 6 xmlns:task="http://www.springframework.org/schema/task" 7 xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" 8 xsi:schemaLocation="http://www.springframework.org/schema/beans 9 http://www.springframework.org/schema/beans/spring-beans-4.0.xsd 10 http://www.springframework.org/schema/mvc 11 http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd 12 http://www.springframework.org/schema/context 13 http://www.springframework.org/schema/context/spring-context-4.0.xsd 14 http://www.springframework.org/schema/aop 15 http://www.springframework.org/schema/aop/spring-aop-4.0.xsd 16 http://www.springframework.org/schema/tx 17 http://www.springframework.org/schema/tx/spring-tx-4.0.xsd 18 http://www.springframework.org/schema/task 19 http://www.springframework.org/schema/task/spring-task-4.0.xsd 20 http://code.alibabatech.com/schema/dubbo 21 http://code.alibabatech.com/schema/dubbo/dubbo.xsd"> 22 23 24  25  26 <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 27  30 <property name="brokerURL" value="tcp://192.168.200.128:61616"/> 31  32 <property name="userName" value="admin"></property> 33 <property name="password" value="admin"></property> 34 </bean> 35 36  37 <bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean"> 38  39 <property name="connectionFactory" ref="activeMQConnectionFactory"></property> 40  41 <property name="maxConnections" value="5"></property> 42 </bean> 43 44  45 <bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> 46  47 <property name="targetConnectionFactory" ref="pooledConnectionFactoryBean"></property> 48 </bean> 49 50  51 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 52  53 <property name="connectionFactory" ref="singleConnectionFactory"></property> 54  55 <property name="defaultDestinationName" value="pId"/> 56 </bean> 57 58  59 <bean id="customMessageListener" class="cn.itcast.core.service.message.CustomMessageListener"/> 60 61  62 <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 63  64 <property name="connectionFactory" ref="singleConnectionFactory"/> 65  66 <property name="destinationName" value="pId"/> 67  68 <property name="messageListener" ref="customMessageListener"></property> 69 </bean> 70 </beans>

我們來說下和上面配置不同的地方, 我們在這里配置了一個監聽器, 因為接收到 JMS 生產者發過來的消息后我們需要有個監聽器去監聽且將監聽到的消息拿過來處理.
接下來看看監聽器的處理方法做了些什么事情:?
CustomMessageListener.java:

1 /* 2 * 接收MQ中的消息 3 */ 4 public class CustomMessageListener implements MessageListener{ 5 @Autowired 6 private SearchService searchService; 7 8 @Override 9 public void onMessage(Message message) { 10 //先將接收到的消息強轉為ActiveMQ類型的消息 11 //因為在消息發送方那邊傳遞的是Text類型的消息對象, 所以需要轉成ActiveMQTextMessage 12 ActiveMQTextMessage amtm = (ActiveMQTextMessage)message; 13 try { 14 String id = amtm.getText(); 15 System.out.println("接收到的ID:"+id); 16 searchService.insertProductToSolr(Long.parseLong(id)); 17 } catch (JMSException e) { 18 // TODO Auto-generated catch block 19 e.printStackTrace(); 20 } 21 }

因為我們接收到的是string類型的文本, 所以這里我們直接將接收到的消息轉換為ActiveMQText類型, 然后通過getText去得到傳遞過來的id, 然后我們就可以通過這個productId去做相應的操作了.?

接下來就看保存商品信息到Solr服務器的邏輯:
SearchServiceImpl.java:

1 //保存商品信息到Solr服務器中, 通過ActiveMQ 2 public void insertProductToSolr(Long productId){ 3 //TODO 保存商品信息到Solr服務器 4 SolrInputDocument doc = new SolrInputDocument(); 5 //ID 6 doc.setField("id", productId); 7 //名稱 8 Product p = productDao.selectByPrimaryKey(productId); 9 doc.setField("name_ik", p.getName()); 10 //圖片URL 11 doc.setField("url", p.getImgUrls()[0]); 12 //品牌 ID 13 doc.setField("brandId", p.getBrandId()); 14 //價格 sql查詢語句: select price from bbs_sku where product_id = ? order by price asc limit 1 15 SkuQuery skuQuery = new SkuQuery(); 16 skuQuery.createCriteria().andProductIdEqualTo(productId); 17 skuQuery.setOrderByClause("price asc"); 18 skuQuery.setPageNo(1); 19 skuQuery.setPageSize(1); 20 List<Sku> skus = skuDao.selectByExample(skuQuery); 21 doc.setField("price", skus.get(0).getPrice()); 22 //...時間等剩下的省略 23 24 try { 25 solrServer.add(doc); 26 solrServer.commit(); 27 } catch (Exception e) { 28 // TODO Auto-generated catch block 29 e.printStackTrace(); 30 } 31 }

這樣就比較明朗了, ActiveMQ 隊列就是這樣來實現的.?

====================接下來還會有 ActiveMQ 訂閱者模式的示例, 這里只是生產者發送消息給單個消費者, 下次還會更新生產者發送消息給多個消費者.

?2016/09/04 20:32 更新
上面已經說了消息的隊列模式, 及點對點發送消息, 那么接下來就來說下消息的一對多模式, 也就是發布/訂閱模式.
項目原型: 當商品上架后(babasport-product), 發送消息id給solr(babasport-solr)來將商品信息保存到solr服務器和cms(babasport-cms)來對商品詳情頁面做頁面靜態化.

===================
babasport-product:
結構圖:

babasport-product下的項目結構圖:

ProductServiceImpl.java中的上架:

1 @Autowired 2 private JmsTemplate jmsTemplate; 3 4 //上架 5 public void isShow(Long[] ids){ 6 Product product = new Product(); 7 product.setIsShow(true); 8 for (final Long id : ids) { 9 //上下架狀態 10 product.setId(id); 11 productDao.updateByPrimaryKeySelective(product); 12 13 //發送商品ID到ActiveMQ即可. 14 jmsTemplate.send(new MessageCreator() { 15 16 @Override 17 public Message createMessage(Session session) throws JMSException { 18 19 return session.createTextMessage(String.valueOf(id)); 20 } 21 }); 22 } 23 } View Code

mq.xml:

這里面的最大的變化就是將消息發布模式改為了: publish subject.

============================================
babasport-solr:

mq.xml配置文件:

1 <beans xmlns="http://www.springframework.org/schema/beans" 2 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:mvc="http://www.springframework.org/schema/mvc" 3 xmlns:context="http://www.springframework.org/schema/context" 4 xmlns:aop="http://www.springframework.org/schema/aop" 5 xmlns:tx="http://www.springframework.org/schema/tx" 6 xmlns:task="http://www.springframework.org/schema/task" 7 xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" 8 xsi:schemaLocation="http://www.springframework.org/schema/beans 9 http://www.springframework.org/schema/beans/spring-beans-4.0.xsd 10 http://www.springframework.org/schema/mvc 11 http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd 12 http://www.springframework.org/schema/context 13 http://www.springframework.org/schema/context/spring-context-4.0.xsd 14 http://www.springframework.org/schema/aop 15 http://www.springframework.org/schema/aop/spring-aop-4.0.xsd 16 http://www.springframework.org/schema/tx 17 http://www.springframework.org/schema/tx/spring-tx-4.0.xsd 18 http://www.springframework.org/schema/task 19 http://www.springframework.org/schema/task/spring-task-4.0.xsd 20 http://code.alibabatech.com/schema/dubbo 21 http://code.alibabatech.com/schema/dubbo/dubbo.xsd"> 22 23 24  25  26 <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 27  30 <property name="brokerURL" value="tcp://192.168.200.128:61616"/> 31  32 <property name="userName" value="admin"></property> 33 <property name="password" value="admin"></property> 34 </bean> 35 36  37 <bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean"> 38  39 <property name="connectionFactory" ref="activeMQConnectionFactory"></property> 40  41 <property name="maxConnections" value="5"></property> 42 </bean> 43 44  45 <bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> 46  47 <property name="targetConnectionFactory" ref="pooledConnectionFactoryBean"></property> 48 </bean> 49 50  51 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 52  53 <property name="connectionFactory" ref="singleConnectionFactory"></property> 54  55 <property name="defaultDestinationName" value="pId"/> 56 </bean> 57 58  59 <bean id="customMessageListener" class="cn.itcast.core.service.message.CustomMessageListener"/> 60 61  62 <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 63  64 <property name="connectionFactory" ref="singleConnectionFactory"/> 65  66 <property name="destinationName" value="pId"/> 67  68 <property name="messageListener" ref="customMessageListener"></property> 69  70 <property name="pubSubDomain" value="true"/> 71 </bean> 72 </beans> View Code

SearchServiceImpl.java: 保存商品信息到Solr服務器中, 通過ActiveMQ

CustomMessageListener.java: 監聽ActiveMQ中傳遞過來的消息, 且對傳遞過來的消息進行處理:

1 public class CustomMessageListener implements MessageListener{ 2 @Autowired 3 private SearchService searchService; 4 5 @Override 6 public void onMessage(Message message) { 7 //先將接收到的消息強轉為ActiveMQ類型的消息 8 //因為在消息發送方那邊傳遞的是Text類型的消息對象, 所以需要轉成ActiveMQTextMessage 9 ActiveMQTextMessage amtm = (ActiveMQTextMessage)message; 10 try { 11 String id = amtm.getText(); 12 System.out.println("接收到的ID:"+id); 13 searchService.insertProductToSolr(Long.parseLong(id)); 14 } catch (JMSException e) { 15 // TODO Auto-generated catch block 16 e.printStackTrace(); 17 } 18 } 19 } View Code

===============================
babasport-cms:
mq.xml:

1  2  3 <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 4  7 <property name="brokerURL" value="tcp://192.168.200.128:61616"/> 8  9 <property name="userName" value="admin"></property> 10 <property name="password" value="admin"></property> 11 </bean> 12 13  14 <bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean"> 15  16 <property name="connectionFactory" ref="activeMQConnectionFactory"></property> 17  18 <property name="maxConnections" value="5"></property> 19 </bean> 20 21  22 <bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> 23  24 <property name="targetConnectionFactory" ref="pooledConnectionFactoryBean"></property> 25 </bean> 26 27  28 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 29  30 <property name="connectionFactory" ref="singleConnectionFactory"></property> 31  32 <property name="defaultDestinationName" value="pId"/> 33 </bean> 34 35  36 <bean id="customMessageListener" class="cn.itcast.core.service.message.CustomMessageListener"/> 37 38  39 <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 40  41 <property name="connectionFactory" ref="singleConnectionFactory"/> 42  43 <property name="destinationName" value="pId"/> 44  45 <property name="messageListener" ref="customMessageListener"></property> 46  47 <property name="pubSubDomain" value="true"/> 48 </bean> View Code

CustomMessageListener.java: 監聽ActiveMQ中傳遞過來的消息, 且對傳遞過來的消息進行處理:

1 public class CustomMessageListener implements MessageListener{ 2 @Autowired 3 private StaticPageService staticPageService; 4 @Autowired 5 private CMSService cmsService; 6 7 @Override 8 public void onMessage(Message message) { 9 //先將接收到的消息強轉為ActiveMQ類型的消息 10 //因為在消息發送方那邊傳遞的是Text類型的消息對象, 所以需要轉成ActiveMQTextMessage 11 ActiveMQTextMessage amtm = (ActiveMQTextMessage)message; 12 try { 13 String id = amtm.getText(); 14 System.out.println("CMS接收到的ID:"+id); 15 Map<String, Object> root = new HashMap<String, Object>(); 16 17 Product product = cmsService.selectProductById(Long.parseLong(id)); 18 List<Sku> skus = cmsService.selectSkuListByProductIdWithStock(Long.parseLong(id)); 19 //去掉重復的顏色 20 Set<Color> colors = new HashSet<Color>(); 21 for (Sku sku : skus) { 22 colors.add(sku.getColor()); 23 } 24 root.put("colors", colors); 25 root.put("product", product); 26 root.put("skus", skus); 27 28 staticPageService.index(root, id); 29 } catch (JMSException e) { 30 // TODO Auto-generated catch block 31 e.printStackTrace(); 32 } 33 } 34 } View Code

StaticPageServiceImpl.java: 靜態化頁面的核心類:

1 public class StaticPageServiceImpl implements StaticPageService, ServletContextAware{ 2 //SpringMvc 管理 conf 3 private Configuration conf; 4 public void setFreeMarkerConfig(FreeMarkerConfig freeMarkerConfig) { 5 this.conf = freeMarkerConfig.getConfiguration(); 6 } 7 8 //靜態化頁面的方法 9 public void index(Map<String, Object> root, String id){ 10 //輸出目錄: 通過getPath方法獲取的是絕對路徑 11 String path = getPath("/html/product/" + id +".html"); 12 File f = new File(path); 13 File parentFile = f.getParentFile(); 14 if(!parentFile.exists()){ 15 parentFile.mkdirs(); 16 } 17 18 //spring中已經設置了模板路徑:<property name="templateLoaderPath" value="/WEB-INF/ftl/" /> 19 Writer out = null; 20 21 try { 22 //讀 23 Template template = conf.getTemplate("product.html"); 24 25 //設置輸出的位置 26 //寫 27 out = new OutputStreamWriter(new FileOutputStream(f), "UTF-8"); 28 template.process(root, out); 29 } catch (Exception e) { 30 // TODO Auto-generated catch block 31 e.printStackTrace(); 32 }finally { 33 if (out != null) 34 { 35 try { 36 out.close(); 37 } catch (IOException e) { 38 // TODO Auto-generated catch block 39 e.printStackTrace(); 40 } 41 } 42 43 } 44 45 } 46 47 //獲取webapp下的html文件夾所在的位置 48 //將相對路徑轉換為絕對路徑 49 public String getPath(String path){ 50 return servletContext.getRealPath(path); 51 } 52 53 private ServletContext servletContext; 54 @Override 55 public void setServletContext(ServletContext servletContext) { 56 this.servletContext = servletContext; 57 } 58 } View Code

Spring管理 freemarkerConfig配置類:

1  2 <bean class="cn.itcast.core.service.StaticPageServiceImpl"> 3 <property name="freeMarkerConfig"> 4 <bean class="org.springframework.web.servlet.view.freemarker.FreeMarkerConfigurer"> 5  6 <property name="templateLoaderPath" value="/WEB-INF/ftl/" /> 7  8 <property name="defaultEncoding" value="UTF-8"></property> 9 </bean> 10 </property> 11 </bean> View Code

更多關于freemarker的講解請關注我以后的博客...
關于ActiveMQ的內容就更新到這么多.

轉載于:https://www.cnblogs.com/wang-meng/p/5831538.html

總結

以上是生活随笔為你收集整理的ActiveMQ的介绍及使用实例.的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： python类class基础
下一篇： [转]密码技术－实现数字信封和数字签名