Python对接ActiveMQ消息队列_python链接jms activemq

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

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

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

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

安装命令: 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 要和发送消息的保持一致，不然还是会收不到消息。