ActiveMQ

ゝ一世哀愁。 2022-05-21 02:23 161阅读 0赞

> 此为博主(yjclsx)原创文章，如若转载请标明出处，谢谢！  
> \#一、JMS  
> 当前CORBA、DCOM、RMI等RPC中间件技术已广泛应用于各个领域，但是面对规模和复杂度都越来越高的分布式系统，这些技术也显示出其局限性：  
> （1）同步通信：客户发出调用后，必须等待服务对象完成处理并返回结果后才能继续执行；  
> （2）客户和服务对象的生命周期紧密耦合：客户进程和服务对象进程都必须正常运行；如果由于服务对象崩溃或者网络故障导致客户的请求不可达，客户会接收到异常；  
> （3）点对点通信：客户的一次调用只发送给某个单独的目标对象。  
> 面向消息的中间件（Message Oriented Middleware，MOM）较好的解决了以上问题。发送者将消息发送给消息服务器，消息服务器将消息存放在若干队列中，在合适的时候再将消息转发给接收者。这种模式下，发送和接收是异步的，发送者无需等待；二者的生命周期未必相同：发送消息的时候接收者不一定运行，接收消息的时候发送者不一定运行；一对多通信：对于一个消息可以有多个接收者。  
> JAVA消息服务（JMS）定义了Java中访问消息中间件的接口。JMS只是接口，并没有给予实现，实现JMS接口的消息中间件称为JMS Provider，已有的MOM系统包括Apache的ActiveMQ、阿里巴巴的RocketMQ、IBM的MQSeries、Microsoft的MSMQ、BEA的MessageQ、RabbitMQ等待，他们基本都遵循JMS规范。  
> JMS术语：  
> Provider（MessageProvider）：生产者；  
> Consumer（MessageConsumer）：消费者；  
> PTP：Point to Point，即点对点的消息模型；  
> Pub/Sub：Publish/Subscribe，即发布/订阅的消息模型；  
> Queue：队列目标；  
> Topic：主题目标；  
> ConnectionFactory：连接工厂，JMS用它创建连接；  
> Connection：JMS客户端到JMS Provider的链接；  
> Destination：消息的目的地；  
> Session：会话，一个发送或者接收消息的线程，表示一个单线程的上下文，用于发送和接收消息。由于会话是单线程的，所以消息是连续的，就是说消息是按照发送的顺序一个一个接收的。会话的好处是它支持事务，如果用户选择了事务支持，会话上下文将保存一组消息，直到事务被提交才发送这些消息。在提交事务之前，用户可以使用回滚操作取消这些消息。一个会话允许用户创建消息生产者来发送消息，创建消息消费者来接收消息。  
> Message接口（消息）：是在消费者和生产者之间传送的对象，一个消息有三个主要部分，1、消息头（必须）：包括用于识别和为消息寻找路由的操作设置；2、一组消息属性（可选）：包含额外的属性，支持其他提供者和用户的兼容，可以创建定制的字段和过滤器（消息选择器）；3、一个消息体（可选）：允许用户创建五种类型的消息（文本消息、映射消息、字节消息、流消息和对象消息）。消息接口非常灵活，并提供了许多方式来定制消息的内容。  
> JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。  
> ·StreamMessage：Java原始值的数据流  
> ·MapMessage：一套名称-值对  
> ·TextMessage：一个字符串对象  
> ·ObjectMessage：一个序列化的Java对象  
> ·BytesMessage：一个未解释字节的数据流  
> \#二、ActiveMQ  
> ActiveMQ是Apache出品、最流行的、能力强劲的、开源的消息总线。  
> ActiveMQ是一个完全支持JMS1.1和J2EE1.4规模的JMS Provider实现。80%以上的业务我们使用ActiveMQ已经足够满足需求，当然后续如天猫、淘宝网这种大型的电商网站，尤其是双十一这种特殊时间，ActiveMQ需要进行很复杂的优化源码以及架构设计才能完成，我们之后会学习一个更强大的分布式消息中间件：RocketMQ，可以说ActiveMQ是核心，是基础，所以我们必须要掌握好。  
> 下面写一个ActiveMQ的HelloWorld示例，初步了解ActiveMQ的使用。  
> \#\#1、PTP模型的消息生产者

//建立工厂，需要填入用户名、密码和连接的地址端口
    		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
    				ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
    		//通过工厂创建一个connection连接，连接默认是关闭的
    		Connection connection = connectionFactory.createConnection();
    		//开启连接
    		connection.start();
    		//通过connection创建session会话（上下文环境对象），用于接收或发送消息，参数1为是否启用事务，参数2为签收模式
    		Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    		//在PTP模式中，Destination即代表Queue队列，在Pub/Sub模式中代表Topic主题，在程序中可以使用多个Queue和Topic
    		Destination destination = session.createQueue("queue1");
    		//通过session对象创建消息的发送和接收对象（生产者和消费者）：MessageProducer/MessageConsumer
    		MessageProducer messageProducer = session.createProducer(destination);
    		//设置持久化特性，默认是PERSISTENT持久化的，如果容忍消息丢失，可以设置为NON_PERSISTENT来改善性能和减少存储的开销
    		messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    		//创建消息并发送，默认情况下，消息永不过期，除非你设置了消息的过期时间
    		for(int i=1;i<=10;i++){
    			TextMessage textMessage = session.createTextMessage();
    			textMessage.setText("消息"+i);
    			//参数1是消息，参数2是持久化特性，参数3是消息优先级，分0~9级，越大优先级越高，参数4是消息过期时间，单位是毫秒
      messageProducer.send(textMessage, DeliveryMode.NON_PERSISTENT, i>=10?9:i, 1000*60);
    			System.out.println("生产者生成了消息"+i);
    		}
    		//最后一定不要忘了关闭连接
    		if(connection!=null){
    			connection.close();
    		}

\#\#2、PTP模型的消息消费者

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
    				ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
    		Connection connection = connectionFactory.createConnection();
    		connection.start();
    		Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    		Destination destination = session.createQueue("queue1");
    		MessageConsumer messageConsumer = session.createConsumer(destination);
    		//接收消息并确认签收
    		while(true){
       //messageConsumer.receive()方法当没消息可以接收时会进入阻塞状态
    			TextMessage msg = (TextMessage)messageConsumer.receive();
    			if(msg==null) break;
    			msg.acknowledge();
    			System.out.println(msg.getText());
    		}
    		if(connection!=null){
    			connection.close();
    		}

接收消息除了使用阻塞方法receive外，也可以使用MessageListener来监听，这样就不需要使用while循环了，而且不会造成阻塞，主线程可以继续执行，但是注意在接收到所有消息之前，主线程不能把connection连接关闭掉，如下：

messageConsumer.setMessageListener(new MessageListener() {
    			@Override
    			public void onMessage(Message arg0) {
    				try {
    					if(arg0!=null){
    						arg0.acknowledge();
    						System.out.println("消费者接收到消息："+((TextMessage)arg0).getText());
    					}
    				} catch (JMSException e) {
    					e.printStackTrace();
    				}
    			}
    		});

在创建session时需要指定签收模式，签收模式有三种，AUTO\_ACKNOWLEDGE：自动签收；CLIENT\_ACKNOWLEDGE：消费者通过调用消息的acknowledge方法签收消息，实际工作中最常用，因为往往只有当正确处理了消息后才会签收消息，这个模式下签收一个已消费的消息会自动地签收这个session所有已消费消息的收条；DUPS\_OK\_ACKNOWLEDGE：session不必确保传送消息的签收，它可能引起消息的重复，但是降低了session的开销，所以只要客户端能容忍重复的消息时才可使用。  
以上是PTP的消息模型，假如生产者发送了10条消息，如果只有一个消费者那肯定都给它了，如果有多个消费者会平均分配，比如3个消费者，一个4条另外两个3条。  
还有一种消息模型就是Pub/Sub发布订阅消息模型，在这种消息模型下，假如生产者发送了10条消息，无论有几个消费者，只要在发送前订阅了该Topic主题，那他们都可以收到这10条消息，这种模型适合一对多的场景。Pub/Sub模型的生产者和消费者代码与PTP模型的代码一模一样，只需把Destination destination = session.createQueue(“queue1”)改为Destination destination = session.createTopic(“topic1”)就可以了。  
ActiveMQ的web管理界面：[http://127.0.0.1:8161/admin，初始密码是admin/admin，ActiveMQ的管理台是用jetty部署的，所以修改密码到conf目录下的jetty-realm.properties里进行配置。][http_127.0.0.1_8161_admin_admin_admin_ActiveMQ_jetty_conf_jetty-realm.properties]  
ActiveMQ最常用的配置文件是conf目录下的activemq.xml和jetty.xml。  
\#三、ActiveMQ高可用  
ActiveMQ 5.9之后，增加了基于ZooKeeper + LevelDB的Master-Slave实现方式来作为HA（High Availability 高可用性）方案，当然其他两种方式目录共享和数据库共享也依然存在。  
\#\#(1)基于共享文件系统（KahaDB，默认）：

\#\#(2)基于 JDBC：

\#\#(3)基于可复制的LevelDB（推荐采用这种方式）：  
LevelDB是Google开源的持久化KV单机数据库，具有很高的随机写，顺序读/写性能，但是随机读的性能很一般，也就是说，LevelDB很适合应用在查询较少，而写很多的场景。它不包含网络服务封装，LevelDB自己也声明，使用者应该封装自己的网络服务器。ActiveMQ中自带了LevelDB，在activemq根目录/data/leveldb下。

我这里zookeeper服务就一个，但为了zookeeper的高可用性，往往zookeeper也会使用Master-Slave方式，所以至少也有三个zookeeper服务，那在zkAddress=""里配置多个zookeeper服务地址端口，中间用英文逗号隔开即可。  
ActiveMQ的客户端只能访问Master节点的ActiveMQ服务，其他处于Slave节点的ActiveMQ无法访问，当主节点的ActiveMQ挂了，zookeeper就会把其中一个从节点升为主节点，所以客户端连接时应该使用failover协议(失败转移协议)：failover:(tcp://192.168.246.130:61616,tcp://192.168.246.131:61616,tcp://192.168.246.132:61616)?Randomize=false  
当节点数量不足总数量一半时，将不能提供高可用服务，所有节点将无法访问。比如我这里有3个ActiveMQ，1个主2个从，挂一个，可以正常运行，挂两个就不能对外提供服务了。  
\#四、ActiveMQ集群（负载均衡）  
ActiveMQ的集群（或者说是负载均衡Load Balance 简称LB），推荐使用网络连接模式(network connector)。简单的说，就是通过把多个不同的broker实例连接在一起，作为一个整体对外提供服务，从而提高整体对外的消息服务能力。通过这种方式连接在一起的broker实例之间，可以共享队列和消费者列表，从而达到分布式队列的目的。  
在每个activemq主从的activemq.xml配置文件里的broker节点内添加除它之外的其他activemq主从服务的连接信息：  
主从服务1链接主从服务2：

主从服务2链接主从服务1：

如果不止两个，就在networkConnectors下配多个networkConnector。

> 此为博主(yjclsx)原创文章，如若转载请标明出处，谢谢！

[http_127.0.0.1_8161_admin_admin_admin_ActiveMQ_jetty_conf_jetty-realm.properties]: http://127.0.0.1:8161/admin%EF%BC%8C%E5%88%9D%E5%A7%8B%E5%AF%86%E7%A0%81%E6%98%AFadmin/admin%EF%BC%8CActiveMQ%E7%9A%84%E7%AE%A1%E7%90%86%E5%8F%B0%E6%98%AF%E7%94%A8jetty%E9%83%A8%E7%BD%B2%E7%9A%84%EF%BC%8C%E6%89%80%E4%BB%A5%E4%BF%AE%E6%94%B9%E5%AF%86%E7%A0%81%E5%88%B0conf%E7%9B%AE%E5%BD%95%E4%B8%8B%E7%9A%84jetty-realm.properties%E9%87%8C%E8%BF%9B%E8%A1%8C%E9%85%8D%E7%BD%AE%E3%80%82