免费视频淫片aa毛片_日韩高清在线亚洲专区vr_日韩大片免费观看视频播放_亚洲欧美国产精品完整版

打開APP
userphoto
未登錄

開通VIP，暢享免費電子書等14項超值服

開通VIP

首頁

好書

留言交流

下載APP

聯(lián)系客服
ActiveMQ使用線程池實現(xiàn)消息的生產(chǎn)與消費

1。  首先先引入相關(guān)的lib包，重點需引用activemq-client-5.8.0.jar，activemq-core-5.7.0.jar，activemq-pool-5.8.0.jar，activemq-protobuf-1.1.jar等包，其他包

自行配置。


2。  一些公共工具類的代碼：

JMSProducer.java

  1. package com.ffcs.icity.jms;  
  2.   
  3. import java.util.Map;  
  4. import java.util.Set;  
  5. import java.util.concurrent.ExecutorService;  
  6. import java.util.concurrent.Executors;  
  7.   
  8. import javax.jms.Connection;  
  9. import javax.jms.DeliveryMode;  
  10. import javax.jms.Destination;  
  11. import javax.jms.ExceptionListener;  
  12. import javax.jms.JMSException;  
  13. import javax.jms.MapMessage;  
  14. import javax.jms.Message;  
  15. import javax.jms.MessageProducer;  
  16. import javax.jms.Session;  
  17.   
  18. import org.apache.activemq.ActiveMQConnectionFactory;  
  19. import org.apache.activemq.pool.PooledConnectionFactory;  
  20.   
  21. /** 
  22.  * JMS消息生產(chǎn)者 
  23.  * @author linwei 
  24.  * 
  25.  */  
  26. public class JMSProducer implements ExceptionListener{  
  27.       
  28.     //設(shè)置連接的最大連接數(shù)  
  29.     public final static int DEFAULT_MAX_CONNECTIONS=5;  
  30.     private int maxConnections = DEFAULT_MAX_CONNECTIONS;  
  31.     //設(shè)置每個連接中使用的最大活動會話數(shù)  
  32.     private int maximumActiveSessionPerConnection = DEFAULT_MAXIMUM_ACTIVE_SESSION_PER_CONNECTION;  
  33.     public final static int DEFAULT_MAXIMUM_ACTIVE_SESSION_PER_CONNECTION=300;  
  34.     //線程池數(shù)量  
  35.     private int threadPoolSize = DEFAULT_THREAD_POOL_SIZE;  
  36.     public final static int DEFAULT_THREAD_POOL_SIZE=50;  
  37.     //強制使用同步返回數(shù)據(jù)的格式  
  38.     private boolean useAsyncSendForJMS = DEFAULT_USE_ASYNC_SEND_FOR_JMS;  
  39.     public final static boolean DEFAULT_USE_ASYNC_SEND_FOR_JMS=true;  
  40.     //是否持久化消息  
  41.     private boolean isPersistent = DEFAULT_IS_PERSISTENT;  
  42.     public final static boolean DEFAULT_IS_PERSISTENT=true;   
  43.       
  44.     //連接地址  
  45.     private String brokerUrl;  
  46.   
  47.     private String userName;  
  48.   
  49.     private String password;  
  50.   
  51.     private ExecutorService threadPool;  
  52.   
  53.     private PooledConnectionFactory connectionFactory;  
  54.   
  55.     public JMSProducer(String brokerUrl, String userName, String password) {  
  56.         this(brokerUrl, userName, password, DEFAULT_MAX_CONNECTIONS, DEFAULT_MAXIMUM_ACTIVE_SESSION_PER_CONNECTION, DEFAULT_THREAD_POOL_SIZE, DEFAULT_USE_ASYNC_SEND_FOR_JMS, DEFAULT_IS_PERSISTENT);  
  57.     }  
  58.       
  59.     public JMSProducer(String brokerUrl, String userName, String password, int maxConnections, int maximumActiveSessionPerConnection, int threadPoolSize,boolean useAsyncSendForJMS, boolean isPersistent) {  
  60.         this.useAsyncSendForJMS = useAsyncSendForJMS;  
  61.         this.isPersistent = isPersistent;  
  62.         this.brokerUrl = brokerUrl;  
  63.         this.userName = userName;  
  64.         this.password = password;  
  65.         this.maxConnections = maxConnections;  
  66.         this.maximumActiveSessionPerConnection = maximumActiveSessionPerConnection;  
  67.         this.threadPoolSize = threadPoolSize;  
  68.         init();  
  69.     }  
  70.         
  71.     private void init() {  
  72.         //設(shè)置JAVA線程池  
  73.         this.threadPool = Executors.newFixedThreadPool(this.threadPoolSize);  
  74.         //ActiveMQ的連接工廠  
  75.         ActiveMQConnectionFactory actualConnectionFactory = new ActiveMQConnectionFactory(this.userName, this.password, this.brokerUrl);  
  76.         actualConnectionFactory.setUseAsyncSend(this.useAsyncSendForJMS);  
  77.         //Active中的連接池工廠  
  78.         this.connectionFactory = new PooledConnectionFactory(actualConnectionFactory);  
  79.         this.connectionFactory.setCreateConnectionOnStartup(true);  
  80.         this.connectionFactory.setMaxConnections(this.maxConnections);  
  81.         this.connectionFactory.setMaximumActiveSessionPerConnection(this.maximumActiveSessionPerConnection);  
  82.     }  
  83.       
  84.       
  85.     /** 
  86.      * 執(zhí)行發(fā)送消息的具體方法 
  87.      * @param queue 
  88.      * @param map 
  89.      */  
  90.     public void send(final String queue, final Map<String, Object> map) {  
  91.         //直接使用線程池來執(zhí)行具體的調(diào)用  
  92.         this.threadPool.execute(new Runnable(){  
  93.             @Override  
  94.             public void run() {  
  95.                 try {  
  96.                     sendMsg(queue,map);  
  97.                 } catch (Exception e) {  
  98.                     e.printStackTrace();  
  99.                 }  
  100.             }  
  101.         });  
  102.     }  
  103.       
  104.     /** 
  105.      * 真正的執(zhí)行消息發(fā)送 
  106.      * @param queue 
  107.      * @param map 
  108.      * @throws Exception 
  109.      */  
  110.     private void sendMsg(String queue, Map<String, Object> map) throws Exception {  
  111.           
  112.         Connection connection = null;  
  113.         Session session = null;  
  114.         try {  
  115.             //從連接池工廠中獲取一個連接  
  116.             connection = this.connectionFactory.createConnection();  
  117.             /*createSession(boolean transacted,int acknowledgeMode) 
  118.               transacted - indicates whether the session is transacted acknowledgeMode - indicates whether the consumer or the client  
  119.               will acknowledge any messages it receives; ignored if the session is transacted.  
  120.               Legal values are Session.AUTO_ACKNOWLEDGE, Session.CLIENT_ACKNOWLEDGE, and Session.DUPS_OK_ACKNOWLEDGE. 
  121.             */  
  122.             //false 參數(shù)表示 為非事務型消息，后面的參數(shù)表示消息的確認類型  
  123.             session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);  
  124.             //Destination is superinterface of Queue  
  125.             //PTP消息方式       
  126.             Destination destination = session.createQueue(queue);  
  127.             //Creates a MessageProducer to send messages to the specified destination  
  128.             MessageProducer producer = session.createProducer(destination);  
  129.             //set delevery mode  
  130.             producer.setDeliveryMode(this.isPersistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);  
  131.             //map convert to javax message  
  132.             Message message = getMessage(session, map);  
  133.             producer.send(message);  
  134.         } finally {  
  135.             closeSession(session);  
  136.             closeConnection(connection);  
  137.         }  
  138.     }  
  139.       
  140.     private Message getMessage(Session session, Map<String, Object> map) throws JMSException {  
  141.         MapMessage message = session.createMapMessage();  
  142.         if (map != null && !map.isEmpty()) {  
  143.             Set<String> keys = map.keySet();  
  144.             for (String key : keys) {  
  145.                 message.setObject(key, map.get(key));  
  146.             }  
  147.         }  
  148.         return message;  
  149.     }  
  150.       
  151.     private void closeSession(Session session) {  
  152.         try {  
  153.             if (session != null) {  
  154.                 session.close();  
  155.             }  
  156.         } catch (Exception e) {  
  157.             e.printStackTrace();  
  158.         }  
  159.     }  
  160.   
  161.     private void closeConnection(Connection connection) {  
  162.         try {  
  163.             if (connection != null) {  
  164.                 connection.close();  
  165.             }  
  166.         } catch (Exception e) {  
  167.             e.printStackTrace();  
  168.         }  
  169.     }  
  170.       
  171.     @Override  
  172.     public void onException(JMSException e) {  
  173.         e.printStackTrace();  
  174.     }  
  175.   
  176. }  

JMSConsumer.java

  1. package com.ffcs.icity.jms;  
  2.   
  3. import javax.jms.Connection;  
  4. import javax.jms.ConnectionFactory;  
  5. import javax.jms.Destination;  
  6. import javax.jms.ExceptionListener;  
  7. import javax.jms.JMSException;  
  8. import javax.jms.MessageConsumer;  
  9. import javax.jms.MessageListener;  
  10. import javax.jms.Session;  
  11.   
  12. import org.apache.activemq.ActiveMQConnection;  
  13. import org.apache.activemq.ActiveMQConnectionFactory;  
  14. import org.apache.activemq.ActiveMQPrefetchPolicy;  
  15.   
  16.   
  17. /** 
  18.  * JMS消息消費者 
  19.  * @author linwei 
  20.  * 
  21.  */  
  22. public class JMSConsumer implements ExceptionListener {  
  23.   
  24.     //隊列預取策略  
  25.     private int queuePrefetch=DEFAULT_QUEUE_PREFETCH;  
  26.     public final static int DEFAULT_QUEUE_PREFETCH=10;  
  27.       
  28.     private String brokerUrl;  
  29.       
  30.     private String userName;  
  31.   
  32.     private String password;  
  33.   
  34.     private MessageListener messageListener;  
  35.       
  36.     private Connection connection;  
  37.       
  38.     private Session session;  
  39.     //隊列名  
  40.     private String queue;  
  41.       
  42.       
  43.     /** 
  44.      * 執(zhí)行消息獲取的操作 
  45.      * @throws Exception 
  46.      */  
  47.     public void start() throws Exception {  
  48.         //ActiveMQ的連接工廠  
  49.         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(this.userName, this.password, this.brokerUrl);  
  50.         connection = connectionFactory.createConnection();  
  51.         //activeMQ預取策略  
  52.         ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();  
  53.         prefetchPolicy.setQueuePrefetch(queuePrefetch);  
  54.         ((ActiveMQConnection) connection).setPrefetchPolicy(prefetchPolicy);  
  55.         connection.setExceptionListener(this);  
  56.         connection.start();  
  57.         //會話采用非事務級別，消息到達機制使用自動通知機制  
  58.         session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);  
  59.         Destination destination = session.createQueue(this.queue);  
  60.         MessageConsumer consumer = session.createConsumer(destination);  
  61.         consumer.setMessageListener(this.messageListener);  
  62.     }  
  63.       
  64.       
  65.     /** 
  66.      * 關(guān)閉連接 
  67.      */  
  68.     public void shutdown(){  
  69.         try {  
  70.             if (session != null) {  
  71.                 session.close();  
  72.                 session=null;  
  73.             }  
  74.             if (connection != null) {  
  75.                 connection.close();  
  76.                 connection=null;  
  77.             }  
  78.         } catch (Exception e) {  
  79.             e.printStackTrace();  
  80.         }  
  81.     }  
  82.       
  83.     @Override  
  84.     public void onException(JMSException e) {  
  85.         e.printStackTrace();  
  86.     }  
  87.   
  88.   
  89.     public String getBrokerUrl() {  
  90.         return brokerUrl;  
  91.     }  
  92.   
  93.   
  94.     public void setBrokerUrl(String brokerUrl) {  
  95.         this.brokerUrl = brokerUrl;  
  96.     }  
  97.   
  98.   
  99.     public String getUserName() {  
  100.         return userName;  
  101.     }  
  102.   
  103.   
  104.     public void setUserName(String userName) {  
  105.         this.userName = userName;  
  106.     }  
  107.   
  108.   
  109.     public String getPassword() {  
  110.         return password;  
  111.     }  
  112.   
  113.   
  114.     public void setPassword(String password) {  
  115.         this.password = password;  
  116.     }  
  117.   
  118.   
  119.     public String getQueue() {  
  120.         return queue;  
  121.     }  
  122.   
  123.   
  124.     public void setQueue(String queue) {  
  125.         this.queue = queue;  
  126.     }  
  127.   
  128.   
  129.     public MessageListener getMessageListener() {  
  130.         return messageListener;  
  131.     }  
  132.   
  133.   
  134.     public void setMessageListener(MessageListener messageListener) {  
  135.         this.messageListener = messageListener;  
  136.     }  
  137.   
  138.   
  139.     public int getQueuePrefetch() {  
  140.         return queuePrefetch;  
  141.     }  
  142.   
  143.   
  144.     public void setQueuePrefetch(int queuePrefetch) {  
  145.         this.queuePrefetch = queuePrefetch;  
  146.     }  
  147.       
  148.       
  149. }  

MessageHandler.java

  1. package com.ffcs.icity.jms;  
  2.   
  3. import javax.jms.Message;  
  4.   
  5.   
  6. /** 
  7.  * 提供消息操作的回調(diào)接口 
  8.  * @author linwei 
  9.  * 
  10.  */  
  11. public interface MessageHandler {  
  12.   
  13.       
  14.     /** 
  15.      * 消息回調(diào)提供的調(diào)用方法 
  16.      * @param message 
  17.      */  
  18.     public void handle(Message message);  
  19. }  

MultiThreadMessageListener.java

  1. package com.ffcs.icity.jms;  
  2.   
  3. import java.util.concurrent.ExecutorService;  
  4.   
  5. import javax.jms.Message;  
  6. import javax.jms.MessageListener;  
  7.   
  8.   
  9. /** 
  10.  * 消息消費者中使用的多線程消息監(jiān)聽服務 
  11.  * @author linwei 
  12.  * 
  13.  */  
  14. public class MultiThreadMessageListener implements MessageListener {  
  15.   
  16.     //默認線程池數(shù)量  
  17.     public final static int DEFAULT_HANDLE_THREAD_POOL=10;  
  18.     //最大的處理線程數(shù).  
  19.     private int maxHandleThreads;  
  20.     //提供消息回調(diào)調(diào)用接口  
  21.     private MessageHandler messageHandler;  
  22.   
  23.     private ExecutorService handleThreadPool;  
  24.       
  25.       
  26.     public MultiThreadMessageListener(MessageHandler messageHandler){  
  27.         this(DEFAULT_HANDLE_THREAD_POOL, messageHandler);  
  28.     }  
  29.       
  30.     public MultiThreadMessageListener(int maxHandleThreads,MessageHandler messageHandler){  
  31.         this.maxHandleThreads=maxHandleThreads;  
  32.         this.messageHandler=messageHandler;  
  33.         //支持阻塞的固定大小的線程池(自行手動創(chuàng)建的)  
  34.         this.handleThreadPool = new FixedAndBlockedThreadPoolExecutor(this.maxHandleThreads);  
  35.     }  
  36.       
  37.       
  38.     /** 
  39.      * 監(jiān)聽程序中自動調(diào)用的方法 
  40.      */  
  41.     @Override  
  42.     public void onMessage(final Message message) {  
  43.         //使用支持阻塞的固定大小的線程池來執(zhí)行操作  
  44.         this.handleThreadPool.execute(new Runnable() {  
  45.             public void run() {  
  46.                 try {  
  47.                     MultiThreadMessageListener.this.messageHandler.handle(message);  
  48.                 } catch (Exception e) {  
  49.                     e.printStackTrace();  
  50.                 }  
  51.             }  
  52.         });  
  53.     }  
  54.   
  55. }  

FixedAndBlockedThreadPoolExecutor.java

  1. package com.ffcs.icity.jms;  
  2.   
  3. import java.util.concurrent.BlockingQueue;  
  4. import java.util.concurrent.LinkedBlockingQueue;  
  5. import java.util.concurrent.ThreadPoolExecutor;  
  6. import java.util.concurrent.TimeUnit;  
  7. import java.util.concurrent.locks.Condition;  
  8. import java.util.concurrent.locks.ReentrantLock;  
  9.   
  10.   
  11. /** 
  12.  * 支持阻塞的固定大小的線程池 
  13.  * @author linwei 
  14.  * 
  15.  */  
  16. public class FixedAndBlockedThreadPoolExecutor extends ThreadPoolExecutor {  
  17.   
  18.       
  19.     //一個可重入的互斥鎖 Lock，它具有與使用 synchronized 方法和語句所訪問的隱式監(jiān)視器鎖相同的一些基本行為和語義，但功能更強大。  
  20.     //使用 lock 塊來調(diào)用 try，在之前/之后的構(gòu)造中  
  21.     private ReentrantLock lock = new ReentrantLock();  
  22.       
  23.     private Condition condition = this.lock.newCondition();  
  24.       
  25.     public FixedAndBlockedThreadPoolExecutor(int size) {  
  26.         super(size, size, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());  
  27.     }  
  28.   
  29.       
  30.     /** 
  31.      * 當線程池中沒有空閑線程時,會掛起此方法的調(diào)用線程.直到線程池中有線程有空閑線程. 
  32.      */  
  33.     @Override  
  34.     public void execute(Runnable command) {  
  35.         //進行同步鎖定  
  36.         this.lock.lock();  
  37.         super.execute(command);  
  38.         try {  
  39.             //如果線程池的數(shù)量已經(jīng)達到最大線程池的數(shù)量,則進行掛起操作  
  40.             if (getPoolSize() == getMaximumPoolSize()) {  
  41.                 this.condition.await();  
  42.             }  
  43.         } catch (InterruptedException e) {  
  44.             e.printStackTrace();  
  45.         } finally {  
  46.             this.lock.unlock();  
  47.         }  
  48.     }  
  49.       
  50.     @Override  
  51.     protected void afterExecute(Runnable r, Throwable t) {  
  52.         super.afterExecute(r, t);  
  53.         try {  
  54.             this.lock.lock();  
  55.             this.condition.signal();  
  56.         } finally {  
  57.             this.lock.unlock();  
  58.         }  
  59.     }  
  60.       
  61.       
  62. }  


3. 調(diào)用例子說明：

生產(chǎn)者調(diào)用代碼，JMSProducerTest.java

  1. package com.ffcs.icity.test;  
  2.   
  3. import java.util.HashMap;  
  4. import java.util.Map;  
  5.   
  6. import com.ffcs.icity.jms.JMSProducer;  
  7.   
  8. public class JMSProducerTest {  
  9.   
  10.       
  11.     public static void main(String[] args) {  
  12.           
  13.         locationTest();  
  14.         System.out.println("over.");  
  15.     }  
  16.       
  17.     private static void locationTest() {  
  18.         //**  JMSProducer 可以設(shè)置成全局的靜態(tài)變量，只需實例化一次即可使用,禁止循環(huán)重復實例化JMSProducer(因為其內(nèi)部存在一個線程池)  
  19.           
  20.         //支持openwire協(xié)議的默認連接為 tcp://localhost:61616，支持 stomp協(xié)議的默認連接為tcp://localhost:61613。   
  21.         //tcp和nio的區(qū)別  
  22.         //nio://localhost:61617 以及 tcp://localhost:61616均可在 activemq.xml配置文件中進行配置  
  23.         JMSProducer producer = new JMSProducer("nio://localhost:61617", "system", "manager");  
  24.         Map<String, Object> map = new HashMap<String, Object>();  
  25.         map.put("id", "1");  
  26.         map.put("name", "sss1113333");  
  27.         map.put("password", "password");  
  28.         producer.send("test", map);  
  29.     }  
  30.       
  31. }  

消費者調(diào)用代碼，JMSConsumerTest.java

  1. package com.ffcs.icity.test;  
  2.   
  3. import javax.jms.MapMessage;  
  4. import javax.jms.Message;  
  5.   
  6. import com.ffcs.icity.jms.JMSConsumer;  
  7. import com.ffcs.icity.jms.MessageHandler;  
  8. import com.ffcs.icity.jms.MultiThreadMessageListener;  
  9.   
  10. public class JMSConsumerTest {  
  11.   
  12.       
  13.     public static void main(String[] args) throws Exception {  
  14.           
  15.         //**  JMSConsumer 可以設(shè)置成全局的靜態(tài)變量，只需實例化一次即可使用,禁止循環(huán)重復實例化JMSConsumer(因為其內(nèi)部存在一個線程池)  
  16.   
  17.         JMSConsumer consumer = new JMSConsumer();  
  18.         consumer.setBrokerUrl("tcp://localhost:61616");  
  19.         consumer.setQueue("test");  
  20.         consumer.setUserName("system");  
  21.         consumer.setPassword("manager");  
  22.         consumer.setQueuePrefetch(500);  
  23.         consumer.setMessageListener(new MultiThreadMessageListener(50,new MessageHandler() {  
  24.             public void handle(Message message) {  
  25.                 try {  
  26.                     System.out.println("name is " + ((MapMessage)message).getString("name"));  
  27.                     Thread.sleep(5000);  
  28.                 } catch (Exception e) {  
  29.                     e.printStackTrace();  
  30.                 }  
  31.             }  
  32.         }));  
  33.         consumer.start();  
  34.           
  35. //      Thread.sleep(5000);  
  36. //      consumer.shutdown();  
  37.           
  38.     }  
  39.       
  40.       
  41. }  





本站僅提供存儲服務，所有內(nèi)容均由用戶發(fā)布，如發(fā)現(xiàn)有害或侵權(quán)內(nèi)容，請點擊舉報。
打開APP，閱讀全文并永久保存 查看更多類似文章
猜你喜歡
類似文章
Activemq和Lingo實現(xiàn)JMS和異步調(diào)用-Java頻道
ActiveMQ 即時通訊服務 淺析（二）
springboot 集成ActiveMQ
Java Websocket實例【項目實戰(zhàn)系列】
深入掌握JMS（七）：DeliveryMode例子
JMS在Spring框架下的應用
更多類似文章 >>
生活服務
分享 收藏 導長圖 關(guān)注 下載文章
綁定賬號成功
后續(xù)可登錄賬號暢享VIP特權(quán)！
如果VIP功能使用有故障，
可點擊這里聯(lián)系客服！

聯(lián)系客服