ActiveMQ消息间件第一章

安全，可靠，接耦合，异步，高吞吐，低延迟

安装

下载官网：activemq.apache.org/components/…
Mac版选择下面gz的链接。

下载完成后解压后，进入apache-activemq-5.16.0/bin/macosx目录（版本号对应自己下载的版本）
启动activemq

./activemq start

在浏览器内输入 http://localhost:8161/ 访问控制台页面。初始账号/密码是admin。 关闭activemq

./activemq stop

重启activemq

./activemq restart

JMS中的角色

Broker

消息服务器，作为server提供消息核心服务，一个activemq实例，相当于一个服务端
点击Manage ActiveMQ broker可以查看当前这台服务器的信息。

MQ作用-结偶

MQ存储有一个目的地（类似mysql的表），然后有两种类型（queue/topic）

• queue : 只可以消费一次
• topic : 可以被多个不同的消费者消费 消费的时候要指定目的地

Queue简单使用案例

首先创建一个maven项目，然后将pom依赖配置到pom.xml文件中

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

创建一个生产者java类文件

package com.world.mq;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息发送者（生产者）
 * 
 * @author daosen
 *
 */

public class Sender {

	public static void main(String[] args) throws JMSException {
		// 1.获取连接工厂
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD,
				"tcp://localhost:61616");

		// 2.获取一个向ActiveMQ的连接
		Connection connection = connectionFactory.createConnection();
		// 3.获取session
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 4.找目的地，获取destination，消费端，也会从这个目的地取消息
		Queue queue = session.createQueue("user");
		// 5.消息创建者
		MessageProducer producer = session.createProducer(queue);
		// consumer -> 消费者
		// producer -> 创建者
		// 6.创建一个消息
		for (int i = 0; i < 100; i  ) {
			TextMessage textMessage = session.createTextMessage("hello"   i);
			// 7.向目的地写入消息
			producer.send(textMessage);
		}
		// 7.关闭连接
		connection.close();

		System.out.println("System exit.");
	}

}

创建一个消费者java类

package com.world.mq;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息发送者（消费者）
 * 
 * @author daosen
 *
 */

public class Receiver {

	public static void main(String[] args) throws JMSException {
		// 1.获取连接工厂
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD,
				"tcp://localhost:61616");

		// 2.获取一个向ActiveMQ的连接
		Connection connection = connectionFactory.createConnection();
		connection.start();
		// 3.获取session
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 4.找目的地，获取destination，消费端，也会从这个目的地取消息
		Destination queue = session.createQueue("user");
		// 5.获取消息
		MessageConsumer consumer = session.createConsumer(queue);
		// consumer -> 消费者
		// producer -> 创建者
		while (true) {
			TextMessage message = (TextMessage) consumer.receive();
			System.out.println("message："   message.getText());
		}
	}
}

测试代码

1. 启动消费者代码
2. 启动生产者代码
3. 查看消费者对应的控制台打印日志。
   可以看出测试成功了，消费者把消息队列的数据取了出来。 然后我们可以从网页上查看Queues的信息

• Number Of Pending Messages：挂起的消息数目
• Number Of Consumers：消费者人数
• Messages Enqueued：消息排队
• Messages Dequeued：消息去排队

JMS消息由三部分构成

消息头

每个消息头字段都有相应的getter和setter方法。

消息属性

如果需要消息头字段以外的值，那么可以使用消息属性。

消息体

JMS定义的消息类型有TextMessage,MapMessage,BytesMessage,StreamMessage,ObjectMessage