ActiveMQ消息特性：延迟和定时消息投递 AMQ_SCHEDULED_DELAY long 延迟投递的时间 AMQ_SCHEDULED_PERIOD long 重复投递的时间间隔 AMQ_SCHEDULED_REPEAT int 重复投递次数 AMQ_SCHEDULED_CRON String Cron表达式 应用场景 最经典的就是当用户注册时，我们就需要用activeMQ来做为中间件，当用户注册后，我门把用户的邮箱号和验证码等信息通过activeMQ的生产端发送到activeMQ的消息队列中，而一旦消息队列中出现了数据，我们的邮件模块通过实时的监控activeMQ的消息队列就能通过消费端获取到这个数据，染回邮件模块就会自行的去对数据进行解析，给用户发送邮件 cmis应用场景：定时的向ActiveMQ队列获取消息,主系统添加用户，子系统监听队列同步 添加用户、修改、激活、禁言 ActiveMQ 消息发送尽量都是异步,系统解耦用消息中间件 是什么 通过高效可靠的消息传递机制进行与平台无关的数据交流 通过提供的消息传递和消息队列模型在分布式环境下提供系统解耦 发送者把消息发送给消息服务器，消息服务器将消息存放在若干队列/主题中，在合适的时候，消息服务器会将消息转发给接受者 在这个过程中，发送和接受是异步的，也就是发送无需等待，而且发送者和接受者的声明周期也没有必然关系 (主题)尤其在发布pub/订阅sub模式下,也可以完成一对多的通信，即让一个消息有多个接受者 特点:发送者个接受者不必了解对方，只需要确认消息 发送者和接受者不必同时在线 能干嘛 异步通信 应用解耦 流量削峰 数据同步 冗余存储 如: 订单系统 ->发送消息 MQ 消费消息 -> 仓储系统 怎么玩 控制消息的消费顺序 配置集群容错的MQ集群 生产端,负责发送消息到交换机; 消费端,监听队列,获取队列的消息,进行消费; 整体架构:连接工厂 获得连接 通过连接得到session 通过session创建消息目的地队列/主题 再通过session创建消息生产者/消费者,再通过session创建消息以生产者发送至MQ、监听接收消息 receive()/receive(4000):区别不见不散一直等/过时不候时间到直接停止消费者 消息监听时,System.in.read()禁止停止服务 消息头：常用如 1.jmsdestination投到哪队列还是主题、 2.jmsdeliverymode持久或非持久 3.jmsexpiration过期时间、 4.jmspriority消息优先级、0-4普通消息，5-9加急消息 5.jmsmessageid消息唯一标识id 消息体: 1.TextMessage文本消息 常用 3.MapMessage映射消息 常用 4.ByteMessage字节消息 5.StreamMessage流消息 2.ObjectMessage对象消息 消息属性: 如果需要除消息头字段以外的值，那么可以使用消息属性 识别/去重/重点标注等操作非常有用的方法:如：setStringProperty("",""); JMS消息可靠性：(分为持久性、事务、签收Acknowledge) ActiveMQ消息持久化:KahaDB、JDBC、LevelDB messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)当MQ服务宕机消息不存在 messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT)当MQ服务宕机消息依然存在 queue 队列默认是持久化的 topic 将connection.start()放置topic后边，为启动持久化的主题 Subscribers发布订阅：先启动订阅再e启动生产 事务偏生产者/签收偏消费者 事务时需commit()异常时回滚,场景：避免消息重复消费 事务消息和非事务消息参数1：事务，2.签收 connection.createSession(Boolean.true,Session.AUTO_ACKNOWLEDGE) connection.createSession(Boolean.false,Session.AUTO_ACKNOWLEDGE) 签收 Session.AUTO_ACKNOWLEDGE自动签收 Session.CLIENT_ACKNOWLEDGE手动签收,客户端调用message.acknowledge()方法手动签收 ActiveMq-嵌入式Broker： 其实就是用代码的形式启动ActiveMQ将MQ嵌入到Java代码中，以便随时用随时启动，在用的时候再去启动这样能节省了资源，也保证了可靠性 用ActiveMQ Broker作为独立的消息服务器来构建Java应用，相当于是一个ActiveMQ服务器的实例 ActiveMQ也支持在vm中通信基于嵌入式的broker,能够无缝的集成其他java应用 引入jar:com.fasterxml.jackson.core BrokerService bs = new BrokerService(); bs.setUseJmx(true); bs.addConnector("tcp://localhost:61616"); bs.start(); ActiveMQ整合Spring 队列生产者 队列消费者 主题生产消费 监听配置 ActiveMQ整合SpringBoot 队列生产者 队列生产者间隔定投 队列消费者 主题生产者 主题消费者 ActiveMQ的传输协议 查看/修改conf/activemq.xml activemq的web窗口 传输协议分为：tcp/openwire、amqp、stomp(关于流)、mqtt(物联网)、ws(webservice)、Nio、 网络协议：tcp、nio、udp(比tcp好、但不具备可靠性)、ssl（安全协议）、https、VM ssl(安全协议) TCP连接的url形式如：tcp://hostname:port?key=value&key=valu后面的参数是可选 Nio连接的url形式如：nio://hostname:port?key=value 生产环境一般为：NIO，Nio比tcp更偏重于底层访问操作 Nio协议的使用场景： 可能有大量的Client去连接到Broker上，一般情况下大量的client去连接Broker(tcp)是被 操作系统的线程所限制的，因此，Nio的实现比tcp需要更少的线程去运行，所以建议使用 nio协议。 演示: <transportConnector name="nio" url="nio://0.0.0.0:61618?trace=true"> 如何使端口即支持NIO网络Io模型，又支持多协议？ 使用auto关键字，auto+nio 使用"+"符号来为端口设置多种特性 ActiveMQ的消息存储和持久化 MQ的高可用：事务、持久、签收(自身)、（物理持久化将MQ文件复制到其他机器磁盘存储） 持久化分为：JDBC(基于第三方数据库)、5.3AMQ（基于日志）、5.4KahaDB(可汗DB) 5.9LeveDB(基于文件的本地数据库存储) 5.9Replicated LevelDB Store（带复制功能） kahadb：基于日志文件， kahadb存储原理：消息存储使用一个事务日期和仅仅用一个索引文件来存储它所有的地址。 KahaDB是一个专门针对消息持久化的解决方案，他对典型的消息使用模式进行了优化 数据被追加到data logs中，当不再需要log文件中的数据的时候，log文件会被丢弃 四类文件一把锁：db-1.log(每满如32M，增加)、db.data(包含索引)，db.free(当前db.data文件那些页面是空闲的,所有空闲ID),db.redo(进行消息恢复)、lock(文件锁、表示获得kahadb读写权限的broker) JDBC存储原理： 1.MQ+Mysql 2.MQ lib添加mysql驱动包 3.activemq.xml配置 <persistenceAdapter> <jdbcPersistenceAdapter dataSource="#mysql-ds" 数据库名称createTablesOnStartup="true">是否在启动的时候创建数据表 </persistenceAdapter> 4.数据库连接池配置(数据源) 5.建仓sql和建表 建名为activemq库 表为：ACTIVEMQ_MSGS (对列表) ACTIVEMQ_ACKS (主题表) ACTIVEMQ_LOCK (锁) 表自动生成，配置过如：createTablesOnStartup="true" 6.代码运行验证 一定要开启：messageProducer.setDeliveryMode(DeliveryMode.PERSISTENCE) 队列:当DeliveryMode设置为NON_PERSISTENOE时，消息保存到内存中 当DeliveryMode设置为PERSISTENCE时,消息保存在broker的相应的文件或者数据库 消息一旦被Consumer消费就从broker(数据库)中同步删除 主题 7.数据库情况 queue:在运行消费看看表activemq_msgs数据变化 topic:先启动消费者订阅再运行生产，activemq_acks 8.小总结 ACTIVEMQ_MSGS (对列表) ACTIVEMQ_ACKS (主题表) ACTIVEMQ_LOCK (锁) 在集群环境中才有用，只有一个Broker可以获得消息，称为Master Broker， 其他的只能作为备份等待Master Broker不可用，在可能成为下一个Master Broker, 这个表用于记录那个Broker是当前的Master Broker 9.常见坑 1.数据库驱动jar包 2.表自动生成，配置过如：createTablesOnStartup="true" 3.下划线坑爹 “Java.lang.IIlegalStateExcption:BeanFactory not initialized or already closed” 这是因为你的操作系统的机器名中有“_”符号，更改机器名并重启解决 JDBC高速缓存配置： ActiveMQ journal:克服JDBC Store的不足 1.<journalPersistenceAdapterFactory journalLogFiles="4" journalLogFileSize="32768" useJournal="true" useQuickJournal="true" dataSource="#mysql-ds" dataDirectory="activemq-data"/> 2.重启配置文件 3.运行 降低mysql压力，写入journal高速缓存 ActiveMQ持久化总结 MQ所在服务器宕机消息不会丢失 ActiveMQ的多节点集群 未学完 zookeeper+replicated-leveldb-store的主从集群,避免单点故障 ActiveMQ的高级特性： 未学完 异步投递 延迟投递 定时投递 消费重试机制 死信队列 防止重复调用