ActiveMQ消息中間件Producer和Consumer

時間  2020-12-23
欄目 ActiveMQ 简体版
原文   原文鏈接


wKiom1XUiqHzUG7LAAMAStjJo7g288.jpg



生產者代碼：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
package  com.java1234.activemq;
 
import  javax.jms.Connection;
import  javax.jms.ConnectionFactory;
import  javax.jms.Destination;
import  javax.jms.JMSException;
import  javax.jms.MessageProducer;
import  javax.jms.Session;
import  javax.jms.TextMessage;
 
import  org.apache.activemq.ActiveMQConnection;
import  org.apache.activemq.ActiveMQConnectionFactory;
 
/**
  * 消息生產者
  * @author Administrator
  *
  */
public  class  JMSProducer {
 
     private  static  final  String USERNAME=ActiveMQConnection.DEFAULT_USER;  // 默認的連接用戶名
     private  static  final  String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;  // 默認的連接密碼
     private  static  final  String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;  // 默認的連接地址
     private  static  final  int  SENDNUM= 10 ;  // 發送的消息數量
     
     public  static  void  main(String[] args) {
         
         ConnectionFactory connectionFactory;  // 連接工廠
         Connection connection =  null ;  // 連接
         Session session;  // 會話 接受或者發送消息的線程
         Destination destination;  // 消息的目的地
         MessageProducer messageProducer;  // 消息生產者
         
         // 實例化連接工廠
         connectionFactory= new  ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
         
         try  {
             connection=connectionFactory.createConnection();  // 通過連接工廠獲取連接
             connection.start();  // 啓動連接
             session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  // 創建Session
             destination=session.createQueue( "FirstQueue1" );  // 創建消息隊列
             messageProducer=session.createProducer(destination);  // 創建消息生產者
             sendMessage(session, messageProducer);  // 發送消息
             session.commit();
         }  catch  (Exception e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }  finally {
             if (connection!= null ){
                 try  {
                     connection.close();
                 }  catch  (JMSException e) {
                     // TODO Auto-generated catch block
                     e.printStackTrace();
                 }
             }
         }
     }
     
     /**
      * 發送消息
      * @param session
      * @param messageProducer
      * @throws Exception
      */
     public  static  void  sendMessage(Session session,MessageProducer messageProducer) throws  Exception{
         for ( int  i= 0 ;i<JMSProducer.SENDNUM;i++){
             TextMessage message=session.createTextMessage( "ActiveMQ 發送的消息" +i);
             System.out.println( "發送消息：" + "ActiveMQ 發送的消息" +i);
             messageProducer.send(message);
         }
     }
}


消費者分兩種：

1.主動接受消息receive()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
package  com.java1234.activemq;
 
import  javax.jms.Connection;
import  javax.jms.ConnectionFactory;
import  javax.jms.Destination;
import  javax.jms.JMSException;
import  javax.jms.MessageConsumer;
import  javax.jms.Session;
import  javax.jms.TextMessage;
 
import  org.apache.activemq.ActiveMQConnection;
import  org.apache.activemq.ActiveMQConnectionFactory;
 
/**
  * 消息消費者
  * @author Administrator
  *
  */
public  class  JMSConsumer {
 
     private  static  final  String USERNAME=ActiveMQConnection.DEFAULT_USER;  // 默認的連接用戶名
     private  static  final  String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;  // 默認的連接密碼
     private  static  final  String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;  // 默認的連接地址
     
     public  static  void  main(String[] args) {
         ConnectionFactory connectionFactory;  // 連接工廠
         Connection connection =  null ;  // 連接
         Session session;  // 會話 接受或者發送消息的線程
         Destination destination;  // 消息的目的地
         MessageConsumer messageConsumer;  // 消息的消費者
         
         // 實例化連接工廠
         connectionFactory= new  ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
                 
         try  {
             connection=connectionFactory.createConnection();   // 通過連接工廠獲取連接
             connection.start();  // 啓動連接
             session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);  // 創建Session
             destination=session.createQueue( "FirstQueue1" );   // 創建連接的消息隊列
             messageConsumer=session.createConsumer(destination);  // 創建消息消費者
             while ( true ){
                 TextMessage textMessage=(TextMessage)messageConsumer.receive( 100000 );
                 if (textMessage!= null ){
                     System.out.println( "收到的消息：" +textMessage.getText());
                 } else {
                     break ;
                 }
             }
         }  catch  (JMSException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         } 
     }
}

2.使用listener被動監聽消息（非阻塞的）（觀察者模式）


Listener代碼

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package  com.java1234.activemq;
 
import  javax.jms.JMSException;
import  javax.jms.Message;
import  javax.jms.MessageListener;
import  javax.jms.TextMessage;
 
/**
  * 消息監聽
  * @author Administrator
  *
  */
public  class  Listener  implements  MessageListener{
 
     @Override
     public  void  onMessage(Message message) {
         // TODO Auto-generated method stub
         try  {
             System.out.println( "收到的消息：" +((TextMessage)message).getText());
         }  catch  (JMSException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }
     }
 
}



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
package  com.java1234.activemq;
 
import  javax.jms.Connection;
import  javax.jms.ConnectionFactory;
import  javax.jms.Destination;
import  javax.jms.JMSException;
import  javax.jms.MessageConsumer;
import  javax.jms.Session;
import  javax.jms.TextMessage;
 
import  org.apache.activemq.ActiveMQConnection;
import  org.apache.activemq.ActiveMQConnectionFactory;
 
/**
  * 消息消費者
  * @author Administrator
  *
  */
public  class  JMSConsumer2 {
 
     private  static  final  String USERNAME=ActiveMQConnection.DEFAULT_USER;  // 默認的連接用戶名
     private  static  final  String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;  // 默認的連接密碼
     private  static  final  String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;  // 默認的連接地址
     
     public  static  void  main(String[] args) {
         ConnectionFactory connectionFactory;  // 連接工廠
         Connection connection =  null ;  // 連接
         Session session;  // 會話 接受或者發送消息的線程
         Destination destination;  // 消息的目的地
         MessageConsumer messageConsumer;  // 消息的消費者
         
         // 實例化連接工廠
         connectionFactory= new  ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);
                 
         try  {
             connection=connectionFactory.createConnection();   // 通過連接工廠獲取連接
             connection.start();  // 啓動連接
             session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);  // 創建Session
             destination=session.createQueue( "FirstQueue1" );   // 創建連接的消息隊列
             messageConsumer=session.createConsumer(destination);  // 創建消息消費者
             messageConsumer.setMessageListener( new  Listener());  // 註冊消息監聽
         }  catch  (JMSException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         } 
     }
}



如果想主動的去接受消息，而不用異步消息監聽的話，把consumer.setMessageListener(this)改爲Message message = consumer.receive()，手動去調用MessageConsumer的receive方法即可。



wKiom1XUiqHzUG7LAAMAStjJo7g288.jpg



生產者代碼：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
package  com.java1234.activemq;
 
import  javax.jms.Connection;
import  javax.jms.ConnectionFactory;
import  javax.jms.Destination;
import  javax.jms.JMSException;
import  javax.jms.MessageProducer;
import  javax.jms.Session;
import  javax.jms.TextMessage;
 
import  org.apache.activemq.ActiveMQConnection;
import  org.apache.activemq.ActiveMQConnectionFactory;
 
/**
  * 消息生產者
  * @author Administrator
  *
  */
public  class  JMSProducer {
 
     private  static  final  String USERNAME=ActiveMQConnection.DEFAULT_USER;  // 默認的連接用戶名
     private  static  final  String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;  // 默認的連接密碼
     private  static  final  String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;  // 默認的連接地址
     private  static  final  int  SENDNUM= 10 ;  // 發送的消息數量
     
     public  static  void  main(String[] args) {
         
         ConnectionFactory connectionFactory;  // 連接工廠
         Connection connection =  null ;  // 連接
         Session session;  // 會話 接受或者發送消息的線程
         Destination destination;  // 消息的目的地
         MessageProducer messageProducer;  // 消息生產者
         
         // 實例化連接工廠
         connectionFactory= new  ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
         
         try  {
             connection=connectionFactory.createConnection();  // 通過連接工廠獲取連接
             connection.start();  // 啓動連接
             session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  // 創建Session
             destination=session.createQueue( "FirstQueue1" );  // 創建消息隊列
             messageProducer=session.createProducer(destination);  // 創建消息生產者
             sendMessage(session, messageProducer);  // 發送消息
             session.commit();
         }  catch  (Exception e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }  finally {
             if (connection!= null ){
                 try  {
                     connection.close();
                 }  catch  (JMSException e) {
                     // TODO Auto-generated catch block
                     e.printStackTrace();
                 }
             }
         }
     }
     
     /**
      * 發送消息
      * @param session
      * @param messageProducer
      * @throws Exception
      */
     public  static  void  sendMessage(Session session,MessageProducer messageProducer) throws  Exception{
         for ( int  i= 0 ;i<JMSProducer.SENDNUM;i++){
             TextMessage message=session.createTextMessage( "ActiveMQ 發送的消息" +i);
             System.out.println( "發送消息：" + "ActiveMQ 發送的消息" +i);
             messageProducer.send(message);
         }
     }
}


消費者分兩種：

1.主動接受消息receive()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
package  com.java1234.activemq;
 
import  javax.jms.Connection;
import  javax.jms.ConnectionFactory;
import  javax.jms.Destination;
import  javax.jms.JMSException;
import  javax.jms.MessageConsumer;
import  javax.jms.Session;
import  javax.jms.TextMessage;
 
import  org.apache.activemq.ActiveMQConnection;
import  org.apache.activemq.ActiveMQConnectionFactory;
 
/**
  * 消息消費者
  * @author Administrator
  *
  */
public  class  JMSConsumer {
 
     private  static  final  String USERNAME=ActiveMQConnection.DEFAULT_USER;  // 默認的連接用戶名
     private  static  final  String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;  // 默認的連接密碼
     private  static  final  String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;  // 默認的連接地址
     
     public  static  void  main(String[] args) {
         ConnectionFactory connectionFactory;  // 連接工廠
         Connection connection =  null ;  // 連接
         Session session;  // 會話 接受或者發送消息的線程
         Destination destination;  // 消息的目的地
         MessageConsumer messageConsumer;  // 消息的消費者
         
         // 實例化連接工廠
         connectionFactory= new  ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
                 
         try  {
             connection=connectionFactory.createConnection();   // 通過連接工廠獲取連接
             connection.start();  // 啓動連接
             session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);  // 創建Session
             destination=session.createQueue( "FirstQueue1" );   // 創建連接的消息隊列
             messageConsumer=session.createConsumer(destination);  // 創建消息消費者
             while ( true ){
                 TextMessage textMessage=(TextMessage)messageConsumer.receive( 100000 );
                 if (textMessage!= null ){
                     System.out.println( "收到的消息：" +textMessage.getText());
                 } else {
                     break ;
                 }
             }
         }  catch  (JMSException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         } 
     }
}

2.使用listener被動監聽消息（非阻塞的）（觀察者模式）


Listener代碼

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package  com.java1234.activemq;
 
import  javax.jms.JMSException;
import  javax.jms.Message;
import  javax.jms.MessageListener;
import  javax.jms.TextMessage;
 
/**
  * 消息監聽
  * @author Administrator
  *
  */
public  class  Listener  implements  MessageListener{
 
     @Override
     public  void  onMessage(Message message) {
         // TODO Auto-generated method stub
         try  {
             System.out.println( "收到的消息：" +((TextMessage)message).getText());
         }  catch  (JMSException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }
     }
 
}



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
package  com.java1234.activemq;
 
import  javax.jms.Connection;
import  javax.jms.ConnectionFactory;
import  javax.jms.Destination;
import  javax.jms.JMSException;
import  javax.jms.MessageConsumer;
import  javax.jms.Session;
import  javax.jms.TextMessage;
 
import  org.apache.activemq.ActiveMQConnection;
import  org.apache.activemq.ActiveMQConnectionFactory;
 
/**
  * 消息消費者
  * @author Administrator
  *
  */
public  class  JMSConsumer2 {
 
     private  static  final  String USERNAME=ActiveMQConnection.DEFAULT_USER;  // 默認的連接用戶名
     private  static  final  String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;  // 默認的連接密碼
     private  static  final  String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;  // 默認的連接地址
     
     public  static  void  main(String[] args) {
         ConnectionFactory connectionFactory;  // 連接工廠
         Connection connection =  null ;  // 連接
         Session session;  // 會話 接受或者發送消息的線程
         Destination destination;  // 消息的目的地
         MessageConsumer messageConsumer;  // 消息的消費者
         
         // 實例化連接工廠
         connectionFactory= new 
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程