Python对接ActiveMQ消息队列

今天药忘吃喽~ 2022-09-15 05:45 214阅读 0赞

activeMQ官方文档:[Using the API — Stomp 7.0.0 documentation][Using the API _ Stomp 7.0.0 documentation]

ActiveMQ支持Python访问，提供了基于STOMP协议（端口为61613）的库。

ActiveMQ的官文[Cross Language Clients][]中给出了更详细的介绍，并附有示例代码，如下图：

第一行为常规Python访问，第二行为使用Jython访问的方式，四个操作。

![ce0d6e3d46ed421383355eec59476f60.png][]

安装命令: pip install setup.py

发送消息到队列中:

import stomp
    
    conn = stomp.Connection()
    conn.set_listener('', MyListener())
    conn.start()
    conn.connect('admin', 'password', wait=True)
    conn.send(body=' '.join(sys.argv[1:]), destination='/queue/test')
    conn.disconnect()

ActiveMQ官网的四个测试：

发送消息到队列Queue属于 点对点模式，不可以重复消费；

发送消息到主题Topic属于 发布/订阅模式，可以重复消费；

# Send a Message to an Apache ActiveMQ Queue
    import stomp
    
    conn = stomp.Connection10()
    
    conn.start()
    
    conn.connect()
    
    conn.send('SampleQueue', 'Simples Assim')
    
    conn.disconnect()
    
    # Receive a Message from an Apache ActiveMQ Queue
    import stomp
    import time
    
    class SampleListener(object):
      def on_message(self, headers, msg):
        print(msg)
    
    conn = stomp.Connection10()
    
    conn.set_listener('SampleListener', SampleListener())
    
    conn.start()
    
    conn.connect()
    
    conn.subscribe('SampleQueue')
    
    time.sleep(1) # secs
    
    conn.disconnect()
    
    # Send a Message to an Apache ActiveMQ Topic
    import stomp
    
    conn = stomp.Connection10()
    
    conn.start()
    
    conn.connect()
    
    conn.send('/topic/SampleTopic', 'Simples Assim')
    
    conn.disconnect()
    
    # Receive a Message from an Apache ActiveMQ Topic （1）
    import stomp
    import time
    
    class SampleListener(object):
      def on_message(self, headers, msg):
        print(msg)
    
    conn = stomp.Connection10()
    
    conn.set_listener('SampleListener', SampleListener())
    
    conn.start()
    
    conn.connect()
    
    conn.subscribe('/topic/SampleTopic')
    
    time.sleep(1) # secs
    
    conn.disconnect()
    
    # Receive a Message from an Apache ActiveMQ Topic （2）
    import stomp
    import time
    
    class SampleListener(object):
      def on_message(self, headers, msg):
        print(msg)
    
    conn = stomp.Connection10()
    
    conn.set_listener('SampleListener', SampleListener())
    
    conn.start()
    
    conn.connect(headers={'client-id':'SampleClient'})
    
    conn.subscribe(destination='/topic/SampleTopic', headers={'activemq.subscriptionName':'SampleSubscription'})
    
    time.sleep(1) # secs
    
    conn.disconnect()

在连接的时候，遇到了第三方发布了一个java代码，他们使用的是tcp协议，坑死我了：

package Com;
    //import org.apache.activemq.ActiveMQConnectionFactory;
    //import javax.jms.*;
    //
    //class Consumer {
    //    public static void main(String[] args) throws JMSException {
    //        //1.创建connectionfacoty
    //        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("账号","密码","tcp://**.***.**.**:****");
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin","admin","tcp://127.0.0.1:61613");
    //        //2.创建connnect,并启动connnect
    //        Connection connection = connectionFactory.createConnection();
    //        connection.start();
    //
    //        //3.创建session,第一个参数是是否使用事务，第二个参数是确认机制
    //        Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
    //        //4.创建目的地【消费者与生产者的目的地相同才能进行消息传递】
            Destination destination = session.createTopic("主题");
    //        Destination destination = session.createTopic("SampleQueue");
    //        //5.创建消费者，第一个参数是目的地，此时创建的消费者要与目的地进行绑定。
    //        MessageConsumer consumer = session.createConsumer(destination);
    //        //6.使用消费者接受消息消息
    //        try {
    //            TextMessage message = (TextMessage) consumer.receive();
    //            System.out.println(message.getText());
    //        }catch (Exception exception){
    //            System.out.println(exception.getMessage());
    //        }
    //        //8.提交事务
    //        //session.commit();
    //        //9.关闭资源
    //        // session.close();
    //        // connection.close();
    //    }
    //}
    
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    class Consumer {
        public static void main(String[] args) throws JMSException {
            //1.创建connectionfacoty
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("thrid","thrid123","tcp://60.191.78.86:8182");
            //2.创建connnect,并启动connnect
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建session,第一个参数是是否使用事务，第二个参数是确认机制
            Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地【消费者与生产者的目的地相同才能进行消息传递】
            Destination destination = session.createTopic("IOT_THIRD_WARN");
            //5.创建消费者，第一个参数是目的地，此时创建的消费者要与目的地进行绑定。
            MessageConsumer consumer = session.createConsumer(destination);
            //6.使用消费者接受消息消息
            while (true){
                TextMessage message = (TextMessage) consumer.receive();
                System.out.println(message.getText());
            }
            //8.提交事务
            //session.commit();
            //9.关闭资源
            // session.close();
            // connection.close();
        }
    }

结果我这边没有发现，所以一直卡住调不通，最后发现他们没有开通stomp协议，也没有开通协议端口，所以就导致java能通，但是python不通的bug，还有一个注意点：这个/topic或者/queue 要和发送消息的保持一致，不然还是会收不到消息。![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAMTAyNOWwj-elng_size_20_color_FFFFFF_t_70_g_se_x_16][]

[Using the API _ Stomp 7.0.0 documentation]: http://jasonrbriggs.github.io/stomp.py/api.html
[Cross Language Clients]: http://activemq.apache.org/cross-language-clients.html
[ce0d6e3d46ed421383355eec59476f60.png]: /images/20220828/9d1548107bbf4a358e99e7536795512e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAMTAyNOWwj-elng_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/3d89841f23d54dec8c854a096f1917e2.png