标签：示例 connection session mq 消息 activemq

activemq

• 前言
• 使用步骤
• • 1.安装
  • 2.java操作activemq
  • • 2.1 队列模式
    • 2.1 主题模式
    • 2.3 topic和queue的对比总结
  • 3.jms
  • 4.发布订阅
  • 5.事务
  • 6.签收
  • 7.broker
  • 8.springboot整合activemq
  • • 8.1 activemq-produce(生产者)
    • 8.2 activemq-consumer(消费者)
    • 8.3 topic模式
  • 9.传输协议
  • • 9.1 nio协议(重点)
    • 9.2 auto+nio
  • 10.activemq存储机制
  • • 10.1 kahaDB消息存储
    • 10.2 jdbc消息存储
    • 10.2 activemq journal
  • 11.zookeeper+replicated+leveldb的主从集群
  • • 11.1 环境搭建
    • 11.2 集群可用性测试
  • 12 高级特性
  • • 12.1 异步投递
    • 12.2 异步投递如何确认发送成功
    • 12.3 延迟投递和定时投递
    • 12.4 消息重试机制
    • 12.5 死信队列
• 总结

前言

mq产品种类

1. kafka
2. rabbitmq
3. rocketmq
4. activemq

优势

1. 能够做到系统解耦,当新的模块接进来时,可以做到代码改动最小;能够解耦
2. 设置流量缓冲池,可以让后端系统按照自身吞吐能力进行消费,不被冲垮;能够削峰
3. 强弱依赖梳理能将非关键调用链路的操作异步化并提升整体系统的吞吐能力;能够异步

使用步骤

1.安装

官网下载

#解压
tar -zxf apache-activemq-5.15.15-bin.tar.gz
#移动
mv apache-activemq-5.15.15 /opt/
#进入
cd /opt/apache-activemq-5.15.15/bin/
#启动
./activemq start

安装好jdk环境,又回来了(建议搞一台虚拟机作为原型,需要环境学习时就克隆一台 )

#停止服务
./activemq stop
#启动服务并将日志写入指定文件(前提文件夹存在)
./activemq start > /opt/activemq-logs/myrunmq.log
#浏览器访问(web访问端口是8161)
http://192.168.59.140:8161/
#默认账户
admin	admin

2.java操作activemq

2.1 队列模式

发送消息（示例）：

    private static final String ACTIVEMQ_URL="tcp://192.168.59.140:61616";
    private static final String QUEUE_NAME="queue01";
    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂,按照指定的url,采用默认的用户名密码
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2.通过连接工厂,获得连接connection
        Connection connection = factory.createConnection();
        connection.start();
        //3.创建会话
        //两个参数,第一个叫事务,第二个叫签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4.创建目的地(具体是队列还是主题,不论队列还是主题都是继承自destination)
        Queue queue = session.createQueue(QUEUE_NAME);
        //5.创建消息生产者
        MessageProducer messageProducer = session.createProducer(queue);
        //6.通过使用messageProducer生产3条消息发送到mq的队列里面
        for (int i = 0; i < 3; i++) {
            //7.创建消息
            TextMessage textMessage = session.createTextMessage("msg---" + (i + 1));
            //8.通过messageProducer发送给mq
            messageProducer.send(textMessage);
        }
        //9.关闭资源
        messageProducer.close();
        session.close();
        connection.close();
        System.out.println("*******消息发布到MQ完成");
    }

接收消息（示例）：

    private static final String ACTIVEMQ_URL="tcp://192.168.59.140:61616";
    private static final String QUEUE_NAME="queue01";
    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂,按照指定的url,采用默认的用户名密码
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2.通过连接工厂,获得连接connection
        Connection connection = factory.createConnection();
        connection.start();
        //3.创建会话
        //两个参数,第一个叫事务,第二个叫签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4.创建目的地(具体是队列还是主题,不论队列还是主题都是继承自destination)
        Queue queue = session.createQueue(QUEUE_NAME);
        //5.创建消费者
        MessageConsumer consumer = session.createConsumer(queue);
        while (true){
            //6.接收消息
            TextMessage textMessage = (TextMessage) consumer.receive();
            if (textMessage != null) {
                System.out.println("********消费者接受到消息:"+textMessage.getText());
            }else{
                break;
            }
        }
        consumer.close();
        session.close();
        connection.close();
        //9.关闭资源
        System.out.println("*******接收消息完成");
    }

通过监听的方式接收消息（示例）：

        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                if (message != null && message instanceof TextMessage) {
                    TextMessage textMessage=(TextMessage)message;
                    try {
                        System.out.println("********消费者监听到的消息是:"+textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        //保证main方法不退出
        System.in.read();
        consumer.close();
        session.close();
        connection.close();

队列模式特点（示例）：

1. 没有时间上的相关性
2. 保存消息,等待消费者消费
3. 如果同时有多名消费者(相同的queue),那么消息会被平均分配

2.1 主题模式

发送消息（示例）：

//目的地变了,由queue变成了topic
Topic topic = session.createTopic(TOPIC_NAME);

接收消息（示例）：

//接收消息也是一样的
Topic topic = session.createTopic(TOPIC_NAME);
MessageConsumer consumer = session.createConsumer(topic);
//当然可以用Lambda表达式简化代码
consumer.setMessageListener((message)->{
    if(message!=null && message instanceof TextMessage){
        TextMessage textMessage= (TextMessage) message;
        try {
            System.out.println("********消费者监听到的消息是:"+textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
});

主题模式特点（示例）：

1. 有时间上的相关性,只能消费自它订阅之后发布的消息
2. 不保存消息,假如无人订阅就去生产,那就是一条废消息,一般先启动消费者再启动生产者
3. 每条消息都会被所有订阅者消费(人人有份)

2.3 topic和queue的对比总结

3.jms

java message service(java消息服务是javaEE中的一个技术)

jms的组成结构和特点（示例）：

• jms provider --> 实现就jms接口和规范的消息中间件,也就是我们的mq服务器

• jms producer --> 消息生产者,创建和发送jms消息的客户端应用

• jms consumer --> 消息消费者,接收和处理jms消息的客户端应用

• jms message --> 重要

  • 消息头

    • jmsdestination --> 目的地
    • jmsdeliverymode --> 是否持久化(默认持久化)
    • jmsexpiration --> 过期时间(默认永不过期)
    • jmspriority --> 消息优先级(0-9十个级别,0-4是普通消息,5-9是加急消息,默认是4级)
    • jmsmessageid -->唯一识别每个消息的标识由mq产生

  • 消息体

    • 封装集体的消息数据
    • 5种消息体格式
      1. textmessage --> 普通字符串消息,包含一个string
      2. mapmessage --> 一个map类型的消息,key作为string类型,而值为java的基本类型
      3. bytesmessage --> 二进制数据消息,包含一个byte[]
      4. streammessage --> java数据流消息,用标准流操作来顺序的填充和
      5. objectmessage --> 对象消息,包含一个可序列化的java对象
    • 发送和接受的消息体类型必须一致

  • 消息属性

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

    • 识别,去重,重点标注等操作

    • textMessage.setStringProperty("chen", "vip");
      

4.发布订阅

发布订阅模式本质其实是持久化的topic

发送消息（示例）：

private static final String ACTIVEMQ_URL="tcp://192.168.59.140:61616";
private static final String TOPIC_NAME="pub-sub";
public static void main(String[] args) throws JMSException {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    Connection connection = factory.createConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic(TOPIC_NAME);
    MessageProducer messageProducer = session.createProducer(topic);
    //持久化
    messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
    //start调用的位置变了
    connection.start();
    for (int i = 0; i < 3; i++) {
        TextMessage textMessage = session.createTextMessage("topic name---" + (i + 1));
        messageProducer.send(textMessage);
    }
    messageProducer.close();
    session.close();
    connection.close();
    System.out.println("*******持久化topic name 消息发布到MQ完成");
}

接收消息（示例）：

private static final String ACTIVEMQ_URL="tcp://192.168.59.140:61616";
private static final String TOPIC_NAME="pub-sub";
public static void main(String[] args) throws JMSException, IOException {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    Connection connection = factory.createConnection();
    //谁订阅
    connection.setClientID("zs");
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic(TOPIC_NAME);
    //持久化topic
    TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "remark ...");
    //start调用的位置变了
    connection.start();
    Message message = topicSubscriber.receive();
    while (message != null) {
        TextMessage textMessage= (TextMessage) message;
        System.out.println("收到的持久化topic: "+textMessage.getText());
        message = topicSubscriber.receive(5000L);
    }
    session.close();
    connection.close();
}

1. 一定先运行一次消费者,等于先mq注册,类似我订阅了这个主题
2. 然后在运行生产者发送消息
3. 此时,无论消费者是否在线,都会接收到,不在线的话,下次连接时会把没收到的消息都接下来

5.事务

事务偏生产者

生产者角度（示例）：

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

1. 当事务为false时,只要执行send,就会进入到队列中
2. 当事务未true时,先执行send在执行commit,消息才被真正的提交的队列中

消费者立场（示例）：

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

1. 当事务为false时,只要执行receive,消息就会被消费
2. 当事务未true时,先执行receive在执行commit,消息才会被真正消费

6.签收

签收偏消费者

非事务手动签收（示例）：

//手动签收
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
//一旦设置成手动签收,需要对每一条消息,进行签收 
textMessage.acknowledge();

有事务手动签收（示例）：

//有事务的手动签收
Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
//情况1:只commit不acknowledge,消息被正常消费,不存在重复消费
session.commit();
//情况2:只acknowledge不commit,消费未被正常消费,存在重复消费
textMessage.acknowledge();

结论:事务大于ack

7.broker

代码如下（示例）：

相当于一个activemq服务器实例
说白了,broker其实就是实现了用代码的形式启动activemq将mq嵌入到java代码中,以便随时用随时启动,在用的时候再去启动这样能节省了资源,也保证了可靠性

8.springboot整合activemq

8.1 activemq-produce(生产者)

引入依赖（示例）：

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

yaml配置（示例）：

server:
  port: 8001
spring:
  activemq:
    broker-url: tcp://192.168.59.140:61616
    user: admin
    password: admin
  jms:
    pub-sub-domain: false #false=queue true=topic
#定义队列名称
myqueue: boot-activemq-queue

配置bean(声明队列)（示例）：

@Configuration
public class ActivemqConfig {
    @Value("${myqueue}")
    private String myqueue;

    @Bean
    public Queue queue(){
        return new ActiveMQQueue(myqueue);
    }
}

编写消息生产者（示例）：

@Component
public class QueueProduce {
    @Autowired
    JmsMessagingTemplate jmsMessagingTemplate;

    @Autowired
    Queue queue;

    public void produceMsg(){
        jmsMessagingTemplate.convertAndSend(queue,"*****"+ UUID.randomUUID().toString());
    }
}

主程序（示例）：

@SpringBootApplication
@EnableJms
public class ActivemqApplication {
    public static void main(String[] args) {
        SpringApplication.run(ActivemqApplication.class, args);
    }
}

测试消息发送（示例）：

@SpringBootTest
class ActivemqApplicationTests {
    @Autowired
    QueueProduce queueProduce;
    @Test
    void contextLoads() {
        queueProduce.produceMsg();
    }
}

8.2 activemq-consumer(消费者)

和produce差不多,修改下端口
删除队列声明和消息发送
新建消费消息类
consumer主程序不需要@EnableJms

@Component
public class QueueConsumer {
    @Autowired
    JmsMessagingTemplate jmsMessagingTemplate;

    @JmsListener(destination = "${myqueue}")
    public void receive(TextMessage textMessage) throws JMSException {
        System.out.println("*******消费者收到的消息是:"+textMessage.getText());
    }
}

8.3 topic模式

topic模式就是把配置文件的pub-sub-domain改成true
声明queue改成声明topic

9.传输协议

9.1 nio协议(重点)

#进入
cd /opt/apache-activemq-5.15.15/conf/
#备份
cp activemq.xml activemq.xml.bak
#修改
vi activemq.xml

<transportConnector name="nio" uri="nio://0.0.0.0:61618?trace=true"/>

修改url连接协议

private static final String ACTIVEMQ_URL="nio://192.168.59.140:61618";

9.2 auto+nio

<transportConnector name="auto+nio" uri="auto+nio://0.0.0.0:61608?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600&amp;org.apache.activemq.transport.nio.SelectorManager.corePoolSize=20&amp;org.apache.activemq.transport.nio.Se1ectorManager.maximumPoo1Size=50"/>

//可以使用nio协议连接
private static final String ACTIVEMQ_URL="nio://192.168.59.140:61608";
//也可以用tcp协议连接
private static final String ACTIVEMQ_URL="tcp://192.168.59.140:61608";

10.activemq存储机制

10.1 kahaDB消息存储

基于日志文件,从activemq5.4开始默认的持久化插件

db-number.log

每32M一个文件,文件名按照数字进行编号,当不再有引用到数据文件中的任何消息时,文件会被删除或归档

db.data

该文件包含了持久化的BTree索引,索引了消息数据记录中的消息,他是消息的索引文件

db.free

当前db.data文件里哪些页面是空闲的,文件具体内容是所有空闲业的id

db.redo

用来进行消息回复,如果kahaDB消息存储在强制退出后启动,用来回复BTree索引

lock

文件锁,表示当前获得kahaDB读写权限的broker

10.2 jdbc消息存储

添加mysql驱动（示例）：

修改配置（示例）：

#修改配置
vi /opt/apache-activemq-5.15.15/conf/activemq.xml
#修改消息存储类型为jdbc
<persistenceAdapter> 
  <jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true"/> 
</persistenceAdapter>
#定义数据源
<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close"> 
  <property name="driverClassName" value="com.mysql.jdbc.Driver"/> 
      <property name="url" value="jdbc:mysql://192.168.59.140/activemq?relaxAutoCommit=true"/> 
      <property name="username" value="root"/> 
      <property name="password" value="root"/> 
      <property name="poolPreparedStatements" value="true"/> 
</bean> 

创建activemq数据库

启动activemq,成功就可以看到表已经建好了

测试发送消息(当消息被消费时则会从数据库删除)

10.2 activemq journal

使用高速缓存写入技术,大大提高了性能
当消费者的消费速度能够及时跟上生产消息的生产速度时,journal文件能够大大减少需要写入到DB中的消息

举个例子:
生产者生产了1000条消息,这1000条消息会保存到journal文件,如果消费者消费速度很快的情况下,在journal文件还没有同步到DB之前,消费者已经消费了90%以上的消息,那么这个时候只需要同步剩余的10%的消息到DB.如果消费者的消费速度很慢,这个时候journal文件可以使消息以批量方式写到DB

#编辑配置文件
vi /opt/apache-activemq-5.15.15/conf/activemq.xml
#修改配置
<persistenceFactory> 
  <journalPersistenceAdapterFactory
  	journalLogFiles="4"
  	journalLogFileSize="32768"
  	useJournal="true"
  	useQuickJournal="true"
  	dataSource="#mysql-ds"
  	dataDirectory="activemq-data"/>
</persistenceFactory>

测试消息发送

11.zookeeper+replicated+leveldb的主从集群

11.1 环境搭建

需要zookeeper集群,zookeeper集群跳转链接

#创建文件夹
mkdir mq-cluster && cd mq-cluster
#递归拷贝,并重命名文件
cp -r /opt/apache-activemq-5.15.15 mq_node01
cp -r /opt/apache-activemq-5.15.15 mq_node02
cp -r /opt/apache-activemq-5.15.15 mq_node03
#编辑hosts
vi /etc/hosts:
#新增名字映射
192.168.59.140 myactivemq
#修改mq_node02和mq_node03(01不动)
vim mq-node02/conf/jetty.xml

修改统一的brokerName(activemq.xml)
vim mq-node01/conf/activemq.xml

修改存储模式为levelDB(activemq.xml)

#编辑配置文件
vim mq-node01/conf/activemq.xml
#修改内容
<persistenceAdapter>
    <replicatedLevelDB
      directory="${activemq.data}/leveldb"
      replicas="3"
      bind="tcp://0.0.0.0:63631"
      zkAddress="localhost:2181,localhost:2182,localhost:2183"
	  sync="local_disk"
      zkPath="/activemq/leveldb-stores"
      hostname="chen-server"/>
  </persistenceAdapter>

修改node02和node03的消息端口(node01默认的就行)

#启动activemq
./mq_node01/bin/activemq start
./mq_node02/bin/activemq start
./mq_node03/bin/activemq start

11.2 集群可用性测试

//修改连接url(故障迁移)
private static final String ACTIVEMQ_URL="failover:(tcp://192.168.59.140:61616,tcp://192.168.59.140:61617,tcp://192.168.59.140:61618)?randomize=false";
private static final String QUEUE_NAME="cluster-queue";

干掉一个activemq节点,它会自动切换到另外一个活着的
测试发送(不贴代码了,用之前的)
故障迁移失败,zookeeper选举了,但是address为null,原因是(我找了一天,就离谱,我用的最新的发行版5.15.15)

更换activemq版本5.13.4

12 高级特性

12.1 异步投递

activemq支持同步和异步两种模式(默认是异步的)
除非指定使用同步,或者在未使用事务的前提下发送持久化消息,这两种情况下都是同步的

同步发送,会阻塞produce直到broker返回一个确认,表示消息已经被安全持久化到磁盘.确认机制提供了消息安全的保障,但同时会阻塞客户端带来了很大的延迟

异步发送,最大化produce端的发送效率,但是不能有效的保证消息发送成功
且需要消耗较多的client短内存同时也会导致broker端性能消耗增加

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//开启异步投递
factory.setUseAsyncSend(true);

12.2 异步投递如何确认发送成功

由于消息不阻塞,生产者会认为所有send的消息均被成功发送至mq
如果mq突然宕机,此时生产者端内存中尚未被发送至mq的消息都会丢失
所以,正确的异步发送方法是需要接收回调的
并由客户端在判断一次是否发送成功

12.3 延迟投递和定时投递

默认没有开启,需要修改activemq.xml文件broker属性(记得重启mq)

12.4 消息重试机制

哪些情况会引起消息重发

1. client开启了事务,且在session中调用了rollback()
2. client开启了事务,并在调用commit()之前关闭或者没有commit
3. client在手动签收模式下,在session中调用了recover()

消息默认重发时间间隔和重发次数
间隔: 1
次数: 6

有毒消息poison ACK
一个消息超过了最大重发次数,消费端会给mq发送一个"poison ack",表示这个消息有毒,告诉broker不要再发了,这个时候broker会把这个消息放到DLQ(死信队列)中

测试–>生产者发送消息,消费者开启事务,但是不commit,重复消费6次(第7次就拿不到消息了)

12.5 死信队列

当一条消息被重发了多次以后(默认6次)将会被mq移入死信队列,开发人员可以在这个队列中查看处理出错的消息,进行人工干预

---

总结

文章主要内容来自B站尚硅谷

标签：示例,connection,session,mq,消息,activemq
来源： https://blog.csdn.net/qq_51138261/article/details/117853520

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。