zoukankan      html  css  js  c++  java
  • 分布式消息通信(ActiveMQ)

    分布式消息通信(ActiveMQ)

    应用场景

    异步通信

    应用解耦

    流量削峰

     

    # ActiveMQ安装

    1. 下载

      http://activemq.apache.org/

    2. 压缩包上传到Linux系统

      apache-activemq-5.15.9-bin.tar.gz

    3. 解压缩

      tar -zxvf apache.activemq-5.15.0-bin.tar.gz

    4. 启动

      bin目录下: ./activemq start

      ####查看是否启动命令:./activemq status

    5. 进入管理后台(默认后台管理端口8161;默认openwire端口61616)

      http://127.0.0.1:8161/admin admin admin

    6. 错误处理

      查看日志:./activemq console

      • hostname不合法

           ERROR | Failed to start Apache ActiveMQ (localhost, ID:VM_0_5_centos-46296-1554189350972-0:1)
       java.net.URISyntaxException: Illegal character in hostname at index 7: ws://VM_0_5_centos:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600
               at java.net.URI$Parser.fail(URI.java:2848)[:1.8.0_201]
               at java.net.URI$Parser.parseHostname(URI.java:3387)[:1.8.0_201]
               at java.net.URI$Parser.parseServer(URI.java:3236)[:1.8.0_201]
               at java.net.URI$Parser.parseAuthority(URI.java:3155)[:1.8.0_201]
               at java.net.URI$Parser.parseHierarchical(URI.java:3097)[:1.8.0_201]
               at java.net.URI$Parser.parse(URI.java:3053)[:1.8.0_201]
               at java.net.URI.<init>(URI.java:673)[:1.8.0_201]
               at

      解决办法:

       编辑文件:vim /etc/hostname,文件内直接填写新的主机名(可以尝试只用这一步)
       
       如果还不行,执行下面的:
       编辑文件:vim /etc/sysconfig/network
       改HOSTNAME=xxxxxxx为 HOSTNAME=wly 保存
       改映射: Vim /etc/hosts
       127.0.0.1 new-hostname.domainname ******
       星号为原来的主机名,现只将*****改成为wly 保存
       
       重启linux机器:reboot

       

    JMS概念和规范

    java消息服务(java message service)

    MOM(message oriented middleware)面向消息中间件

    转载:https://www.cnblogs.com/xinhuaxuan/p/6104274.html

    JMS是什么   JMS 全称:Java Message Service,Java消息服务,是Java EE中的一个技术。

    JMS规范

      JMS定义了Java 中访问消息中间件的接口,并没有给予实现,实现JMS接口的消息中间件成为JMS Provider,例如:Active MQ

    JMS Provider

      实现JMS接口和规范的消息中间件

    JMS message

      JMS的消息,JMS消息由三部分组成:消息头、消息属性、消息体

      消息头包含消息的识别消息和路由消息,消息头包含一些标准的属性如下:

       (1)JMSDestination: 消息发送的目的地,主要是指Queue和Topic,由send方法设置.

       (2)JMSDeliveryMode:传送模式。有两种:持久模式非持久模式。一条持久性的消息应该被传输"一次仅仅一次",这就意味着如果JMS提供者出现故障,该消息并不会丢失,它会在服务器恢复之后再次传递。一条非持久的消息最多会传递一次,这意味着服务器出现故障,该消息将永远丢失。由send方法设置

       (3)JMSExpiration:消息过期时间,等于Destination的send方法中的timeToLive值加上发送时刻的GMT的时间值。如果timeToLive值等于零,则JMSExpiration被设置为零,表示该消息永不过期。如果发送后,在消息过期时间之后消息还没有被发送到目的地,则该消息被清除。由send方法设置

       (4)JMSPriority:消息优先级,从0-9十个级别,0-4是普通消息,5-9是加急消息。JMS不要求JMS Provider严格按照这十个优先级发送消息,但必须保证加急消息要先于普通消息到达,默认是4级。由send方法设置

       (5)JMSMessageID:唯一识别每个消息的标识,由JMS Provider产生。由send方法设置

       (6)JMSTimestamp:一个JMS Provider在调用send()方法时自动设置,它是消息被发送和消费者实际接收的时间差。由客户端设置

       (7)JMSCorrelationID:用来连接到另外一个消息,典型的应用是在回复消息中连接到原消息。在大多数情况下,JMSCorrelationID用于将一条消息标记为对JMSMessageID标示的上一条消息的应答,不过,JMSCorrelationID可以是任何值,不仅仅是JMSMessageID。由客户端设置

       (8)JMSType: 消息类型的标识符,由客户端设置

       (9)JMSReplyTo: 提供本消息回复消息的目的地址,由客户端设置

       (10)JMSRedelivered:如果一个客户端收到一个设置了JMSRedelivered属性的消息,则表示可能客户端曾经在早些时候收到过该消息,但并没有签收(acknowledged)。如果该消息被重新传送,JMSRedelivered=true 否则 JMSRedelivered=flase 。由JMS Provider设置

      消息体,JMS API定义了5种消息体格式,也叫消息类型,可以使用不同形式发送接收数据,并可以兼容现有的消息格式。

        包括:TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage

      消息属性,包含以下三种类型的属性:

        1.应用程序设置和添加的数据,比如:message.setStringProperty("userName",userName);

        2.JMS定义的属性,使用"JMSX"作为属性名的前缀, connection.getMetaData().getJMSXPropertyNames() 方法返回所有连接支持的JMSX属性的名字。

        3.JMS供应商特定的属性

    JMS producer

      消息生产者,创建和发送JMS消息的客户端应用

    JMS consumer

      消息消费者,创建和处理JMS消息的客户端应用

    JMS domains: 消息传递域

      JMS规范中定义了两种消息传递域: 点对点(point-to-point,简写成PTP);消息传递域和发布/订阅消息传递域(publish/subscribe,简写成pub/sub)

      1.点对点消息传递域的特点如下:

       a.每个消息只能有一个消费者

       b.消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态,它都可以提取消息。

      2.发布/订阅消息传递域的特点如下:

        a.每个消息可以有多个消费者

        b.生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消费自它订阅之后发布的消息。JMS规范允许客户创建持久订阅(setClientID),这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。

        

      3.在点对点消息传递域中,目的地被称为队列(queue);在发布/订阅消息传递域中,目的地被称为主题(topic)

     

    Connection factory: 连接工厂,用来创建连接对象,以连接到JMS的provider

    JMS Connection: 封装了客户与JMS提供者之间的一个虚拟的连接 JMS Session: 是生产和消费消息的一个单线程上下文 会话用于创建消息生产者(producer)、消息消费者(consumer)和消息(message)等。会话提供了一个事务性的上下文,在这个上下文中,一组发送和接收被组合到了一个原子操作中。 Destination:消息发送到的目的地 Acknowledge:签收 Transaction:事务 JMS client: 用来收发消息的Java应用

    JMS的可靠性机制:JMS消息之后被确认后,才会认为是被成功消费。

    事务性会话,即设置为ture,消息会在commit后自动确认;

    非事务性会话,即设置为false,在该模式下,消息是否被确认取决于创建会话时的应答模式。:

    AUTO_ACKNOWLEDGE:当客户端成功receive后,消息自动确认

    CLIENT_ACKNOWLEDGE:客户端手动确认(textMessage.acknowledge)

    DUPS_OK_ACKNOWLEDGE:延迟确认(可以设置时间)。

     

    ActiveMQ测试

    编写一个测试类对ActiveMQ进行测试,首先得向pom文件中添加ActiveMQ相关的jar包:

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

    queue的发送代码如下(生产者):

      public void testMQProducerQueue() throws Exception{
             //1、创建工厂连接对象,需要制定ip和端口号
             ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
             //2、使用连接工厂创建一个连接对象
             Connection connection = connectionFactory.createConnection();
             //3、开启连接
             connection.start();
             //4、使用连接对象创建会话(session)对象
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             //5、使用会话对象创建目标对象,包含queue和topic(一对一和一对多)
             Queue queue = session.createQueue("test-queue");
             //6、使用会话对象创建生产者对象
             MessageProducer producer = session.createProducer(queue);
             //7、使用会话对象创建一个消息对象
             TextMessage textMessage = session.createTextMessage("hello!test-queue");
             //8、发送消息
             producer.send(textMessage);
             //9、关闭资源
             producer.close();
             session.close();
             connection.close();
        }

    接收代码(消费者):

         public void TestMQConsumerQueue() throws Exception{
             //1、创建工厂连接对象,需要制定ip和端口号
             ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
             //2、使用连接工厂创建一个连接对象
             Connection connection = connectionFactory.createConnection();
             //3、开启连接
             connection.start();
             //4、使用连接对象创建会话(session)对象
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             //5、使用会话对象创建目标对象,包含queue和topic(一对一和一对多)
             Queue queue = session.createQueue("test-queue");
             //6、使用会话对象创建生产者对象
             MessageConsumer consumer = session.createConsumer(queue);
             //7、向consumer对象中设置一个messageListener对象,用来接收消息
             consumer.setMessageListener(new MessageListener() {
     
                 @Override
                 public void onMessage(Message message) {
                     // TODO Auto-generated method stub
                     if(message instanceof TextMessage){
                         TextMessage textMessage = (TextMessage)message;
                         try {
                             System.out.println(textMessage.getText());
                        } catch (JMSException e) {
                             // TODO Auto-generated catch block
                             e.printStackTrace();
                        }
                    }
                }
            });
             //8、程序等待接收用户消息
             System.in.read();
             //9、关闭资源
             consumer.close();
             session.close();
             connection.close();
        }

    然后当我们运行queue发送的时候可以看到队列里已经有一条消息了,但没有发送出去: 然后在运行queue 的接收端,可以看到消息已经发出了:

     

    broker

    可以本地创建并运行。

     BrokerService brokerservice = new BrokerService();
     brokrService.setUseJmx(true);
     brokrService.addConnector("tcp://localhost:61616");
     brokrService.start();

    FAQ

     Q:消息的发送策略
     A:持久化消息(默认)/非持久化消息
     PS:设置消息发送端发送持久化消息`异步方式`: connectionFactory.setUseAsyncSend(true);
        设置消息发送端发送非持久化消息(默认为异步方式):textMessage.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENCE)
        回执窗口大小设置:connectionFactory.setProducerWindowSize();
        如果需要对非持久化消息的每次发送的消息都获得broker的回执:connectionFactory.setAllwaysSyncSend();
     Q:comsumer获取消息时pull还是push
     A:默认情况下,mq服务器(broker)采用异步方式向客户端主动推送消息(push)
     
        prefetchsize:预取消息数量(broker每次主动推送的消息数量)。Queue默认值是1000;topic持久化消息默认值是100,非持久化消息默认值是32766。如果设置为0,此时对于consumer来说,就是pull模式。
    Q:acknowledge为什么能够在第5次把前面执行的消息确认掉?
    A:源码deliveryMessage:存在之前执行完但没有确认的消息队列
    
    #消息确认
    ACK_TYPE:消息端和broker交换ack指令的时候,还需要告知broker ACK_TYPE
    	REDELIVERED_ACK_TYPE:重发策略
    	DELIVERED_ACK_TYPE:消息已经接受,但是尚未处理结束
    	STANDARD_ACK_TYPE:消息处理成功
    

     

    ActiveMQ+spring整合

    service-jms.xml

    实现MessageListener

    ActiveMQ的传输协议

    client与broker的通信协议

    支持的协议TCP(默认)、UDP、NIO、SSL、Http(s)、vm

    ActiveMQ持久化存储

    1.kahaDB(默认,基于文件的存储方式)

    2.AMQ(基于文件的存储方式)

    写入速度快,容易恢复

    文件默认大小是32M

    3.JDBC(基于数据库的存储)

    <jdbcPersistenceAdapter dataSource="#mysqlDataSource"  createTablesOnStartup="true"/>
    

    连接数据库成功后,会创建三张表ACTIVEMQ_ACKS、ACTIVEMQ_LOCK、ACTIVEMQ_MSGS

    4.Memory(基于内存的存储)

    5.LevelDB

    5.8版本以后引入的持久化策略,通常用于集群配置

     

    ActiveMQ网络连接

    networkConnector:用来配置broker与broker之间的通信连接

    <!--静态网络连接-->
    <networkConnectors>
    	<networkConnectot uri="static://(tcp://192.168.1.1:61616,tcp://192.168.1.2:61616)"/>
    </networkConnectors>
    
    #双向连接:duplex
    
    <!--丢失的消息:配置消息回流,解决该问题-->
    <policyEntry queue=">" enableAudit="false">
    	<networkBridgeFilterFactory>
            <conditionalNetwordBridgeFilterFactory replayWhenNoConsumers="true" />
        </networkBridgeFilterFactory>	
    </policyEntry>
    

    容错连接

     ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://192.168.156.44:61616,tcp://192.168.156.45:61616")";
    
  • 相关阅读:
    停止与暂停线程
    flume日志收集框架
    mysql数据库索引
    junit
    freemarker
    python脚本
    java多线程编程核心技术学习-1
    spring 网站
    [MongoDB]学习笔记--Linux 安装和运行MongoDB
    [Spring MVC]学习笔记--form表单标签的使用
  • 原文地址:https://www.cnblogs.com/kinglead/p/10979774.html
Copyright © 2011-2022 走看看