一、ActiveMQ的介绍。
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。
1、主要特点：
（1）. 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP。
（2）.完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)。
（3）.ActiveMQ可以很容易内嵌到使用Spring的系统里面。
（4）.支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA。
（5）. 支持通过JDBC和journal提供高速的消息持久化。
（6）. 从设计上保证了高性能的集群,客户端-服务器,点对点。
（7） 支持Ajax。
（8）. 支持与Axis的整合。
（9）. 可以很容易得调用内嵌JMS provider,进行测试。
2、ActiveMQ的消息形式。
2.1.消息的传递有两种类型：
（1）点对点：一个生产者和一个消费者一一对应。
（2）发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。
2.2、JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。
StreamMessage -- Java原始值的数据流。
MapMessage--一套名称-值对。
TextMessage--一个字符串对象。
ObjectMessage--一个序列化的 Java对象。
BytesMessage--一个字节的数据流。
二、ActiveMQ的安装。
下载地址：https://activemq.apache.org/
1、安装环境。
1.1、需要JDK。
1.2、安装Linux系统。生产环境都是Linux系统。
2、安装步骤。
2.1、把ActiveMQ 的压缩包上传到Linux系统。
2.2、解压缩 。
2.3、启动。
（1）使用bin目录下的activemq命令启动：[root@localhost bin]# ./activemq start
（2）关闭：[root@localhost bin]# ./activemq stop
（3）查看状态：[root@localhost bin]# ./activemq status
（4）进入管理后台：http://192.168.25.168:8161/admin 用户名：admin 密码：admin
三、ActiveMQ的使用方法。
1、Queue。
1.1、生产者（Product）：生产消息，发送端。在Pom.xml文件中引入依赖，这里使用的是activemq-all-5.11.2.jar。
（1）创建 ConnectionFactory 对象，需要指定服务端ip及端口号。
（2）、使用ConnectionFactory对象创建一个Connection对象。
（3）开启连接，调用Connection对象的 start() 方法。
（4）创建Session，使用Connection对象创建一个Session对象。
（5）使用Session对象创建一个Destination对象（queue、topic），这里创建的是Queue对象。
（6）使用Session对象创建一个Product对象。
（7）创建一个Message对象，创建一个TestMessage对象。
（8）使用Product发送消息。
（9）关闭资源。
@Test
public void testQueueProducer() throws Exception {
// 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
//brokerURL服务器的ip及端口号
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：使用ConnectionFactory对象创建一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步：开启连接，调用Connection对象的start方法。
connection.start();
// 第四步：使用Connection对象创建一个Session对象。
//第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
//第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
//参数：队列的名称。
Queue queue = session.createQueue("test-queue");
// 第六步：使用Session对象创建一个Producer对象。
MessageProducer producer = session.createProducer(queue);
// 第七步：创建一个Message对象，创建一个TextMessage对象。
/*TextMessage message = new ActiveMQTextMessage();
message.setText("hello activeMq,this is my first test.");*/
TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");
// 第八步：使用Producer对象发送消息。
producer.send(textMessage);
// 第九步：关闭资源。
producer.close();
session.close();
connection.close();
}
1.2、消费者（Consumer）：接收消息。
（1）创建一个ConnectionFactory对象。
（2）从ConnectionFactory对象中获取一个Connection对象。
（3）开启连接。调用Connection对象的start方法。
（4）使用Connection对象创建一个Session对象。
（5）使用Session对象创建一个Destination对象，和发送端保持一致Queue，并且队列的名称一致。
（6）使用Session对象创建一个Consumer对象。
（7）接收消息。
（8）打印消息。
（9）关闭资源。
@Test
public void testQueueConsumer() throws Exception {
// 第一步：创建一个ConnectionFactory对象。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：从ConnectionFactory对象中获得一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步：开启连接。调用Connection对象的start方法。
connection.start();
// 第四步：使用Connection对象创建一个Session对象。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
Queue queue = session.createQueue("test-queue");
// 第六步：使用Session对象创建一个Consumer对象。
MessageConsumer consumer = session.createConsumer(queue);
// 第七步：接收消息。
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
String text = null;
//取消息的内容
text = textMessage.getText();
// 第八步：打印消息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//等待键盘输入
System.in.read();
// 第九步：关闭资源
consumer.close();
session.close();
connection.close();
}
2、Topic。
2.1、生产者（Product）。
（1）创建ConnectionFactory对象，需要指定服务端ip及端口号。
（2）使用ConnectionFactory对象创建一个Connection对象。
（3）开启连接。调用Connection对象的 start 方法。
（4）使用Connection对象创建一个Session对象。
（5）使用Session对象创建一个Destination对象（topic、queue），这里创建一个Topic对象。
（6）使用Session对象创建一个Product对象。
（7）创建一个Message，创建一个TextMessage对象。
（8）使用Product对象发送消息。
（9）关闭资源。
@Test
public void testTopicProducer() throws Exception {
// 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
// brokerURL服务器的ip及端口号
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：使用ConnectionFactory对象创建一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步：开启连接，调用Connection对象的start方法。
connection.start();
// 第四步：使用Connection对象创建一个Session对象。
// 第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
// 第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个topic对象。
// 参数：话题的名称。
Topic topic = session.createTopic("test-topic");
// 第六步：使用Session对象创建一个Producer对象。
MessageProducer producer = session.createProducer(topic);
// 第七步：创建一个Message对象，创建一个TextMessage对象。
/*
* TextMessage message = new ActiveMQTextMessage(); message.setText(
* "hello activeMq,this is my first test.");
*/
TextMessage textMessage = session.createTextMessage("hello activeMq,this is my topic test");
// 第八步：使用Producer对象发送消息。
producer.send(textMessage);
// 第九步：关闭资源。
producer.close();
session.close();
connection.close();
}
2.2、消费者（Consumer）：接收消息。
（1）创建一个ConnectionFactory对象。
（2）使用ConnectionFactory对象创建一个Connection对象。
（3）开启连接。调用Connection对象的 start 方法开启连接。
（4）创建Session对象。使用Connection对象创建一个Session对象。
（5）使用Session对象创建一个Destination对象。和发送端保持一个的topic，并且话题的名称保持一致。
（6）使用Session对象创建一个Consumer对象。
（7）接收消息。
（8）打印消息。
（9）关闭资源。
@Test
public void testTopicConsumer() throws Exception {
// 第一步：创建一个ConnectionFactory对象。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步：从ConnectionFactory对象中获得一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步：开启连接。调用Connection对象的start方法。
connection.start();
// 第四步：使用Connection对象创建一个Session对象。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。
Topic topic = session.createTopic("test-topic");
// 第六步：使用Session对象创建一个Consumer对象。
MessageConsumer consumer = session.createConsumer(topic);
// 第七步：接收消息。
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
String text = null;
// 取消息的内容
text = textMessage.getText();
// 第八步：打印消息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
System.out.println("topic的消费端03。。。。。");
// 等待键盘输入
System.in.read();
// 第九步：关闭资源
consumer.close();
session.close();
connection.close();
}
标签：session,ConnectionFactory,对象,创建,消息中间件,Connection,Session,使用,ActiveMQ
From： https://www.cnblogs.com/sfwu/p/16879961.html