ActiveMQ demo

Posted 2021-01-27 chen--biao

Maven 配置文件　　

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

 

 

provider 生产者代码

import static commons.Constants.*;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Provider {
    public static void main(String[] args) {
        ConnectionFactory cf = new ActiveMQConnectionFactory(URL); //创建工厂链接对象
        Connection connection = null;
        Session session  = null;
        Queue queue  = null;
        MessageProducer producer  = null;
        try {
            connection = cf.createConnection(); //使用工厂创建链接
            connection.start(); //开启连接
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //使用连接对象创建会话
            queue = session.createQueue(QUEUE_ONE); //使用会话创建目标对象
            producer = session.createProducer(queue); // 使用会话、目标对象创建生产者对象
            Message message = session.createTextMessage("hello ,i‘m  not good"); //使用会话创建消息对象
            producer.send(message);//发送消息
            System.out.println("send:"+message.toString());
        } catch (JMSException e) {
            e.printStackTrace();
        }finally {
            if(producer != null) {
                try {
                    producer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(session != null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}

 

 

Consumer 消费者

import static commons.Constants.*;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {

    public static void main(String[] args) {
        ConnectionFactory cf = new ActiveMQConnectionFactory(URL);
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;
        try {
            connection = cf.createConnection();
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue(QUEUE_ONE);
            consumer = session.createConsumer(queue);
            //向consumer对象中设置一个messageListener对象，用来接收消息
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    if(message instanceof TextMessage) { //当前测试，仅测试Text类型的消息
                        TextMessage text = (TextMessage)message;
                        try {
                            System.out.println(text.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            System.out.println("consumer : "+System.currentTimeMillis());
            System.in.read();
        }catch(Exception e) {
            e.printStackTrace();
        }finally {
            if(consumer != null) {
                try {
                    consumer.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(session != null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
            if(connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

 

 

Constants 常量类

package commons;

public interface Constants {

    String URL = "tcp://192.168.49.128:61616";
    
    String QUEUE_ONE = "CQC_ONE";
        
}