# mq

**Repository Path**: dlr-sfs_admin/mq

## Basic Information

- **Project Name**: mq
- **Description**: 消息中间件学习
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2019-08-10
- **Last Updated**: 2020-12-19

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# mq
各种消息组件:kafka rabbitmq rocketmq activemq(功能都一样,主学activemq)

1.为什么使用中间件: 可以解耦 消峰 异步
![输入图片说明](https://images.gitee.com/uploads/images/2019/0811/000143_7544e7da_5105108.png "1565452793(1).png")

2.linux安装:
  1.mkdir /usr/tools/mq
  2.解压缩:  tar -zxvf /opt/apache-activemq-5.15.9-bin.tar.gz -C /usr/tools/mq/
  3.修改名称: mv apache-activemq-5.15.9/ activemq
  4.启动:cd /usr/tools/mq/activemq/bin/ ./activemq start/stop
  5.默认端口: 61616
  6.是否启动:ps -ef|grep activemq|grep -v grep

3.控制台访问: 默认端口8161 http://192.168.10.20:8161/ 默认用户名 admin/admin

4.java操作:
  4.1 JMS总体架构:通过 连接工厂 获取 连接 获取 session 创建 消息 目的地
  
  4.2  目的地:点对点队列和发布订阅主题两种模式
  
  代码讲解:
  点对点队列,创建成功
  ![输入图片说明](https://images.gitee.com/uploads/images/2019/0811/235623_5144e9f6_5105108.png "1565538889(1).png")
  消费者:
  1.通过receive()方法进行同步阻塞接受,不设置时间一直等待,设置后没有,过期不侯
  2.通过Listener监听接口来实现 
  
  3.点对点队列消费情况:先启动生产者,在启动多个消费者,那个消费者先消费其余的消费者就不消费了，2.先启动多个消费者,在启动生产者,平均消费
  4.点对点的特点:
  ![点对点的特点](https://images.gitee.com/uploads/images/2019/0812/220036_ea70b548_5105108.png "点对点的特点.png")
  
  5.发布订阅主题
  ![发布订阅主题](https://images.gitee.com/uploads/images/2019/0812/230350_fb7290e2_5105108.png "发布订阅.png")
  6.二者区别
  ![二者区别](https://images.gitee.com/uploads/images/2019/0812/230359_e965c8f4_5105108.png "二者区别.png")
  
5.JMS学习:
  1.是俩个程序之间进行异步通信的API,为标准的消息协议和服务提供一组通用接口,包括创建 发送 读取等.当程序通过JMS进行通讯时,之间并不是直接调用,
    通过JMS规范的消息收发组件进行关联的。
    
  2.元素:JMS容器(上边的四个组件) 生产者 消费者 消息(Message)  
  
  3.消息(Message):消息头 消息属性 消息体组成
  
     消息头主要方法:
       setJMSDestination:设置消息的目的地(队列和主题)
       setJMSDeliveryMode:消息是否持久化
       setJMSExpiration:消息的过期时间
       setJMSPriority:消息的优先级(0-9的级别越来越高) 默认4级
       setJMSMessageID:消息的唯一ID
       
     消息体主要概述:
      共有5种消息格式:(切记生产和消费的消息格式必须一致)
 ![输入图片说明](https://images.gitee.com/uploads/images/2019/0814/232015_404d5409_5105108.png "123.png")
     消息属性概述:主要是对消息进行更细致的区分(设置其他属性)
       
6.消息的可靠性(Important):主要基于消息的持久性 消息事务 消息签收
  1.持久性(基于connection连接)
  DeliveryMode.NON_PERSISTENT非持久化和DeliveryMode.PERSISTENT持久化
   重点:当指定消息为非持久化时,如果mq宕机,在重启消息将不复存在,持久化如果消息没被消费,重启还在(默认是持久化的)
   
   6.1点对点队列持久话只需要设置一下DeliveryMode的值即可(就不写代码了)
   6.2订阅发布主题:非持久化没有意义,先启动订阅在启动发布
   6.3点对点 发布订阅持久化的区别:
      点对点无无需调整启动位置 直接设置持久状态就可以,发布订阅生产者和消费者需要调整启动位置,并且消费者需要使用TopicSubscriber接受消息(看代码persistent_topic_consumer)
  
  2.事务(基于session会话 偏于生产者):Session session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    false:不开启事务默认commit,true:开启事务,消息发送后,session关闭前,必须commit(好处是批量消息时,可以回滚)
    当生产者和消费者开启事务时,生产者不commitmq不会产生消息,消费者不commit程序会消费消息,但mq控制台不显示消费,会*重复消费*
    
  3.签收(偏于消费者):  
    AUTO_ACKNOWLEDGE:自动签收 默认的******
    CLIENT_ACKNOWLEDGE:手动签收****** 
    DUPS_OK_ACKNOWLEDGE:允许重复签收
    SESSION_TRANSACTED:事务签收
    ----------------------不写代码了--------------------------------
    非事务手动签收:消费者显示调用textMessage.acknowledge();手动签收
    事务的手动签收:开启事务了,无论哪种签收模式,都被默认为自动签收(**消费者开启事务但没有提交的手动签收会重复消费**)
   
     1.mq启动时,默认执行/usr/tools/mq/activemq/conf下的activemq.xml    
     
     2.将activemq.xml拷贝一份activemq-1.xml
     
     3.启动时指定配置文件:[root@localhost bin]# ./activemq start xbean:file://usr/tools/mq/activemq/conf/activemq-1.xml
7.按照不同的conf启动activeMQ
  **_Broker相当于activeMQ服务器的实例,正常是在服务器启动MQ,现在将通过代码将MQ嵌套在Java代码中,通过代码启动_**
 
  通过Java代码嵌入式启动: BrokerService

8.整合Spring:
     1.生产者 消费者配置一样
     
     <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
            <property name="connectionFactory">
                <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                    <property name="brokerURL" value="tcp://192.168.10.20:61616"/>
                </bean>
            </property>
            <property name="maxConnections" value="100"/>
         </bean>
     
         <!--点对点队列目的地-->
         <bean id="queue" class="org.apache.activemq.command.ActiveMQQueue">
             <constructor-arg value="spring-queue"/>
         </bean>
         <!--这个是主题目的地，一对多的,仅有两个构造函数一个无参一个带String -->
         <bean id="topic" class="org.apache.activemq.command.ActiveMQTopic">
             <constructor-arg value="spring-topic"/>
         </bean>
         <!--Spring提供的JMS工具类 消息的发送 接受-->
         <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
             <property name="connectionFactory" ref="pooledConnectionFactory"/>
             <property name="defaultDestination" ref="topic"/>
             <property name="messageConverter">
                 <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
             </property>
         </bean>
   
         如果消费者是监听模式的话就不用配置Spring提供的JMS工具类 消息的发送 接受 改为
         <!-- 消息监听容器(也可以说是消费者) -->
             <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
                 <property name="connectionFactory" ref="pooledConnectionFactory"/>
                 <!--这里测试是点对点的，使用发布/订阅的话使用topic-->
                 <property name="destination" ref="topic"/>
                 <property name="messageListener" ref="projectMsgLinster"/>
             </bean>
         
         https://blog.csdn.net/changhenshui1990/article/details/79813240
 
9.SringBoot整合:
  Queue
   一:生产者
     1.写配置类开启mq的注解@EnableJms 2.使用JmsMessagingTemplate操作消息
扩展 3. 定时发送消息:生产者标注@Scheduled 主启动类开启定时任务 @EnableScheduling 
   二:消费者
    配置消费和的类 方法标记@JmsListener(destination = "${queue-name}")监听某个队列即可
    
  Topic:和队列一致
  
10.优化配置及集群:
   
   1.MQ的传输协议:auto vm amqp mqtt tcp(基础)  nio(常用) ssl nio-ssl(关于配置官网)
     如何修改mq的传输协议:/usr/tools/mq/activemq/conf==>vi activemq.xml
           <transportConnectors>
              <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
              <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
              <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
              <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
              <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
              <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
           </transportConnectors>
            name=""协议描述 openwire描述的是tcp协议 openwire就是activemq默认的协议
   端口号的修改(上边)
   tcp 的一些优化方案:http://activemq.apache.org/tcp-transport-reference
   
   NIO 协议为ActiveMQ 提供更好的性能
    适合NIO 使用的场景：
       1  当有大量的Client 连接到Broker 上 ， 使用NIO 比使用 tcp 需要更少的线程数量，所以使用 NIO 
       2  可能对于 Broker 有一个很迟钝的网络传输， NIO 的性能高于 TCP 
       
   2.新增nio协议
             <transportConnectors>
               <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
               <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
               <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
               <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
               <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
               <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
   			   <!--添加nio传输协议-->
   			   <transportConnector name="nio" uri="nio://0.0.0.0:61618?trace=true"/>
           </transportConnectors>
    代码配置及NIO协议优化:
       1.基础整合开代码;
       2.优化:
         <transportConnector name="nio" uri="nio://0.0.0.0:61618?trace=true"/>
         只支持以TCP协议为基础得NIO网络IO模型
         配置NIO网络IO模型多协议:
          <transportConnector name="auto+nio" uri="auto+nio://0.0.0.0:61608?maximumConnections=1000
          &wireFormat.maxFrameSize=104857600&org.apache.activemq.transport.nio.SelectorManager.corelPoolSize=20
          &org.apache.activemq.transport.nio.SelectorManager.maximumPoolSize=50"/>