ActiveMQ 入门例子-蒲公英云

ActiveMQ 入门例子

向右看齐 2022-08-03 05:26 253阅读 0赞

一：新建Maven项目，加入如下依赖

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.11.1</version>
</dependency>

二：

server（嵌入式，无需下载安装activemq）

package com.lala.mq;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.broker.BrokerService;
public class Server 
{
    public static void main(String[] args)throws Exception
    {
        BrokerService broker = new BrokerService();
        broker.addConnector("tcp://127.0.0.1:55551");
        broker.setUseJmx(false);
        broker.start();
        TimeUnit.SECONDS.sleep(9999);
    }
}

Producer（生产者）

package com.lala.mq;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class MyProducer
{
    public static void main(String[] args) throws Exception
    {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://127.0.0.1:55551");
        Connection conn = cf.createConnection();
        conn.start();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("cib");
        MessageProducer producer = session.createProducer(queue);
        TextMessage tm = null;
        for(int i=1;i<=5;i++)
        {
            tm = session.createTextMessage("this is test message and msgid = " + i);
            producer.send(tm);
        }
        session.close();
        conn.close();
    }
}

Consumer（消费者）

package com.lala.mq;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class MyConsumer 
{
    public static void main(String[] args)throws Exception
    {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://127.0.0.1:55551");
        Connection conn = cf.createConnection();
        conn.start();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("cib");
        MessageConsumer consumer = session.createConsumer(queue);
        consumer.setMessageListener((message) -> {
            TextMessage tm = (TextMessage)message;
            try {
                System.out.println("收到消息：" + tm.getText());
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}