ActiveMQ
消息中间件产生的背景
在客户端与服务器进行通讯时.客户端调用后，必须等待服务对象完成处理返回结果才能继续执行。
客户与服务器对象的生命周期紧密耦合,客户进程和服务对象进程都都必须正常运行;如果由于服务对象崩溃或者网络故障导致用户的请求不可达,客户会受到异常
点对点通信: 客户的一次调用只发送给某个单独的目标对象。
为什么要使用消息中间件
什么是MQ：客户端和服务器之间的异步通讯
解决高并发、两种通讯方式、发布订阅、异步通信（无需等待）
1、点对点（队列）–生产者、消费者
生产者：主要向队列发送消息
消费者：主要从队列获取消息
生产者向队列发送消息，如果消费者不在，消息会在队列中缓存
消费者从队列获取消息后，消费成功之后，该消息直接在缓存中被清除。
所以解决高并发就是，消费者将能消费的消息消费掉后，其余的在队列中进行排队等待
什么是JMS？
JMS是java的消息服务，JMS的客户端之间可以通过JMS服务进行异步的消息传输。
消息模型
点对点：生产者 消息队列 消费者 不允许多个消费者–可以提高程序的效率 多线程
应用场景：A用户与B用户发送消息;
发布订阅 ： 生产者 主题 消费者 一对多（一般不用redis）
应用场景：用户注册、订单修改库存、日志存储
消息中间件分类
RabbitMQ
是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。同时实现了一个经纪人(Broker)构架，这意味着消息在发送给客户端时先在中心队列排队。对路由(Routing)，负载均衡(Load balance)或者数据持久化都有很好的支持。
Redis
是一个Key-Value的NoSQL数据库，开发维护很活跃，虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。
| 入队 | 出队 | |||||||
|---|---|---|---|---|---|---|---|---|
| 128B | 512B | 1K | 10K | 128B | 512B | 1K | 10K | |
| Redis | 16088 | 15961 | 17094 | 25 | 15955 | 20449 | 18098 | 9355 |
| RabbitMQ | 10627 | 9916 | 9370 | 2366 | 3219 | 3174 | 2982 | 1588 |
ZeroMQ
号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演了这个服务角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果down机，数据将会丢失。其中，Twitter的Storm中使用ZeroMQ作为数据流的传输。
ActiveMQ
是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。RabbitMQ、ZeroMQ、ActiveMQ均支持常用的多种语言客户端 C++、Java、.Net,、Python、 Php、 Ruby等。
Jafka/Kafka
Kafka是Apache下的一个子项目，是一个高性能跨语言分布式Publish/Subscribe消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现复杂均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制来统一了在线和离线的消息处理，这一点也是本课题所研究系统所看重的。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。
其他一些队列列表HornetQ、Apache Qpid、Sparrow、Starling、Kestrel、Beanstalkd、Amazon SQS就不再一一分析。
实现点对点通讯模式
maven依赖
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
</dependencies>
生产者
public class Producter {
public static void main(String[] args) throws JMSException {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
// JMS 客户端到JMS Provider 的连接
Connection connection = connectionFactory.createConnection();
connection.start();
// Session： 一个发送或接收消息的线程
Session session = connection.createSession(Boolean.falst, Session.AUTO_ACKNOWLEDGE);
// Destination ：消息的目的地;消息发送给谁.
// 获取session注意参数值my-queue是Query的名字
Destination destination = session.createQueue("my-queue");
// MessageProducer：消息生产者
MessageProducer producer = session.createProducer(destination);
// 设置不持久化
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 发送一条消息
for (int i = 1; i <= 5; i++) {
sendMsg(session, producer, i);
}
connection.close();
}
/**
* 在指定的会话上，通过指定的消息生产者发出一条消息
*
* @param session
* 消息会话
* @param producer
* 消息生产者
*/
public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {
// 创建一条文本消息
TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
// 通过消息生产者发出消息
producer.send(message);
}
}
消费者
public class JmsReceiver {
public static void main(String[] args) throws JMSException {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
// JMS 客户端到JMS Provider 的连接
Connection connection = connectionFactory.createConnection();
connection.start();
// Session： 一个发送或接收消息的线程
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
// Destination ：消息的目的地;消息发送给谁.
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
Destination destination = session.createQueue("my-queue");
// 消费者，消息接收者
MessageConsumer consumer = session.createConsumer(destination);
while (true) {
TextMessage message = (TextMessage) consumer.receive();
if (null != message) {
System.out.println("收到消息：" + message.getText());
} else
break;
}
session.close();
connection.close();
}
}
发布订阅
生产者
public class TOPSend {
private static String BROKERURL = "tcp://127.0.0.1:61616";
private static String TOPIC = "my-topic";
public static void main(String[] args) throws JMSException {
start();
}
static public void start() throws JMSException {
System.out.println("生产者已经启动....");
// 创建ActiveMQConnectionFactory 会话工厂
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);
Connection connection = activeMQConnectionFactory.createConnection();
// 启动JMS 连接
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(null);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
send(producer, session);
System.out.println("发送成功!");
connection.close();
}
static public void send(MessageProducer producer, Session session) throws JMSException {
for (int i = 1; i <= 5; i++) {
System.out.println("我是消息" + i);
TextMessage textMessage = session.createTextMessage("我是消息" + i);
Destination destination = session.createTopic(TOPIC);
producer.send(destination, textMessage);
}
}
}
消费者
public class TopReceiver {
private static String BROKERURL = "tcp://127.0.0.1:61616";
private static String TOPIC = "my-topic";
public static void main(String[] args) throws JMSException {
start();
}
static public void start() throws JMSException {
System.out.println("消费点启动...");
// 创建ActiveMQConnectionFactory 会话工厂
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);
Connection connection = activeMQConnectionFactory.createConnection();
// 启动JMS 连接
connection.start();
// 不开消息启事物，消息主要发送消费者,则表示消息已经签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建一个队列
Topic topic = session.createTopic(TOPIC);
MessageConsumer consumer = session.createConsumer(topic);
// consumer.setMessageListener(new MsgListener());
while (true) {
TextMessage textMessage = (TextMessage) consumer.receive();
if (textMessage != null) {
System.out.println("接受到消息:" + textMessage.getText());
// textMessage.acknowledge();// 手动签收
// session.commit();
} else {
break;
}
}
connection.close();
}
}
ActiveMQ 持久化
producer.setDeliveryMode(DeliveryMode.PERSISTENT)这句话解决问题
默认是没有持久化机制的；
JMS消息可靠机制
ActiveMQ消息签收机制：
客戶端成功接收一条消息的标志是一条消息被签收，成功应答。
消息的签收情形分两种：
1、带事务的session
如果session带有事务，并且事务成功提交，则消息被自动签收。如果事务回滚，则消息会被再次传送。
2、不带事务的session
不带事务的session的签收方式，取决于session的配置。
Activemq支持一下三種模式：
Session.AUTO_ACKNOWLEDGE 消息自动签收 没有事务 不安全
Session.CLIENT_ACKNOWLEDGE 客戶端调用acknowledge方法手动签收 MQ里还是有消息没有被消费
textMessage.acknowledge();//手动签收
Session.DUPS_OK_ACKNOWLEDGE 不是必须签收，消息可能会重复发送。在第二次重新传送消息的时候，消息 网络延迟情况下会产生MQ幂等性问题
只有在被确认之后，才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。 在事务性会话中，当一个事务被提交的时候，确认自动发生。在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值：
场景1
生产者不开启session，客户端必须有手动签收模式
Session session = createConnection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
消费者不开启session，客户端必须有手动签收模式
TextMessage textMessage = (TextMessage) createConsumer.receive();
//接受消息
textMessage.acknowledge();
场景2
生产者不开启session，客户端自动签收模式
Session session = createConnection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
消费者不开启session，自动签收消息
Session session = createConnection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
场景4
事物消息 生产者以事物形式，必须要将消息提交事物，才可以提交到队列中。
Session session = createConnection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
session.commit();
消费者
Session session = createConnection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
session.commit();
RocketMQ的可靠机制是返回一个对象
使用消息中间注意事项
\1. 消费者代码不要jdbc抛出异常，否则activqmq默认有重试机制。
\2. 如果代码发生异常，需要发布版本才可以解决的问题，不要使用重试机制，采用日志记录方式，定时Job进行补偿。try catch
\3. 如果不需要发布版本解决的问题，可以采用重试机制进行补偿。
消费者如果保证消息幂等性，不被重复消费。
产生原因:网络延迟传输中，会造成进行MQ重试中，在重试过程中，可能会造成重复消费。
解决办法:
1.使用全局MessageID 判断消费方使用同一个，解决幂等性。
2.使用JMS可靠消息机制
消费者集群（不是MQ集群）
MQ自动处理幂等性问题，MQ知道有多少个链接，分批发送
