PHP接入ActiveMQ技巧_ActiveMQ消息中心件介绍

在客户端与做事器进行通讯时.客户端调用后，必须等待做事工具完成处理返回结果才能连续实行。

客户与做事器工具的生命周期紧密耦合,客户进程和做事工具进程都都必须正常运行;如果由于做事工具崩溃或者网络故障导致用户的要求不可达,客户会受到非常

点对点通信: 客户的一次调用只发送给某个单独的目标工具。

(画图演示)

1.2 什么是中间件

面向的中间件(MessageOrlented MiddlewareMOM)较好的办理了以上问

题。
发送者将发送给做事器，做事器将消感存放在若千行列步队中，在得当

的时候再将转发给吸收者。

这种模式下，发送和吸收是异步的，发送者无需等

待; 二者的生命周期未必相同: 发送的时候吸收者不一定运行，吸收的时候

发送者也不一定运行;一对多通信: 对付一个可以有多个吸收者。

2.1 什么是JMS？

JMS是java的做事，JMS的客户端之间可以通过JMS做事进行异步的传输。

2.2 什么是模型

○ Point-to-Point(P2P) --- 点对点○ Publish/Subscribe(Pub/Sub)--- 发布订阅

即点对点和发布订阅模型

2.2.1 P2P (点对点)

P2P

1.P2P模式图

2.涉及到的观点

1.行列步队（Queue）

2.发送者(Sender)

3.吸收者(Receiver)

4.每个都被发送到一个特定的行列步队，吸收者从行列步队中获取消息。
行列步队保留着，直到他们被消费或超时。

3.P2P的特点

1.每个只有一个消费者（Consumer）(即一旦被消费，就不再在行列步队中)

2.发送者和吸收者之间在韶光上没有依赖性，也便是说当发送者发送了之后，不管吸收者有没有正在运行，它不会影响到被发送到行列步队

3.吸收者在成功吸收之后需向行列步队应答成功

如果你希望发送的每个都该当被成功处理的话，那么你须要P2P模式。

运用处景

A用户与B用户发送

2.2.2Pub/Sub (发布与订阅)

Pub/Sub模式图

涉及到的观点

主题（Topic）

发布者（Publisher）

订阅者（Subscriber）

客户端将发送到主题。
多个发布者将发送到Topic,系统将这些通报给多个订阅者。

Pub/Sub的特点

每个可以有多个消费者

发布者和订阅者之间有韶光上的依赖性。
针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的，而且为了消费，订阅者必须保持运行的状态。

为了缓和这样严格的韶光干系性，JMS许可订阅者创建一个可持久化的订阅。
这样，纵然订阅者没有被激活（运行），它也能吸收到发布者的。

如果你希望发送的可以不被做任何处理、或者被一个者处理、或者可以被多个消费者处理的话，那么可以采取Pub/Sub模型

的消费

在JMS中，的产生和是异步的。
对付消费来说，JMS的者可以通过两种办法来消费。

○ 同步

订阅者或吸收者调用receive方法来吸收，receive方法在能够吸收到之前（或超时之前）将一贯壅塞

○ 异步

订阅者或吸收者可以注册为一个监听器。
当到达之后，系统自动调用监听器的onMessage方法。

运用处景:

用户注册、订单修正库存、日志存储

画图演示

RabbitMQ

是利用Erlang编写的一个开源的行列步队，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正是如此，使的它变的非常重量级，更适宜于企业级的开拓。
同时实现了一个经纪人(Broker)构架，这意味着在发送给客户端时先在中央行列步队排队。
对路由(Routing)，负载均衡(Load balance)或者数据持久化都有很好的支持。

Redis

是一个Key-Value的NoSQL数据库，开拓掩护很生动，虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，以是完备可以当做一个轻量级的行列步队做事来利用。
对付RabbitMQ和Redis的入队和出队操作，各实行100万次，每10万次记录一次实行韶光。
测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。
实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。

ZeroMQ

号称最快的行列步队系统，尤其针对大吞吐量的需求场景。
ZMQ能够实现RabbitMQ不善于的高等/繁芜的行列步队，但是开拓职员须要自己组合多种技能框架，技能上的繁芜度是对这MQ能够运用成功的寻衅。
ZeroMQ具有一个独特的非中间件的模式，你不须要安装和运行一个做事器或中间件，由于你的运用程序将扮演了这个做事角色。
你只须要大略的引用ZeroMQ程序库，可以利用NuGet安装，然后你就可以愉快的在运用程序之间发送了。
但是ZeroMQ仅供应非持久性的行列步队，也便是说如果down机，数据将会丢失。
个中，Twitter的Storm中利用ZeroMQ作为数据流的传输。

ActiveMQ

是Apache下的一个子项目。
类似于ZeroMQ，它能够以代理人和点对点的技能实现行列步队。
同时类似于RabbitMQ，它少量代码就可以高效地实现高等运用处景。
RabbitMQ、ZeroMQ、ActiveMQ均支持常用的多种措辞客户端 C++、Java、.Net,、Python、 Php、 Ruby等。

Jafka/Kafka

Kafka是Apache下的一个子项目，是一个高性能跨措辞分布式Publish/Subscribe行列步队系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。
具有以下特性：快速持久化，可以在O(1)的系统开销下进行持久化；高吞吐，在一台普通的做事器上既可以达到10W/s的吞吐速率；完备的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现繁芜均衡；支持Hadoop数据并行加载，对付像Hadoop的一样的日志数据和离线剖析系统，但又哀求实时处理的限定，这是一个可行的办理方案。
Kafka通过Hadoop的并行加载机制来统一了在线和离线的处理，这一点也是本课题所研究系统所看重的。
Apache Kafka相对付ActiveMQ是一个非常轻量级的系统，除了性能非常好之外，还是一个事情良好的分布式系统。

其他一些行列步队列表HornetQ、Apache Qpid、Sparrow、Starling、Kestrel、Beanstalkd、Amazon SQS就不再逐一剖析。

4.1 、window下 ActiveMQ安装

ActiveMQ支配实在很大略，和所有Java一样，要跑java程序就必须先安装JDK并配置好环境变量，这个很大略。

然后解压下载的apache-activemq-5.10-20140603.133406-78-bin.zip压缩包到一个目录，得到解压后的目录构造如下图：

进入bin目录，创造有win32和win64两个文件夹，这2个文件夹分别对应windows32位和windows64位操作系统的启动脚本。

我的实验环境是windowsXP，就进入win32目录，会看到如下目录构造。

个中activemq.bat便是启动脚本，双击启动。

ActiveMQ默认启动到8161端口，启动完了后在浏览器地址栏输入：http://localhost:8161/admin哀求输入用户名密码，默认用户名密码为admin、admin，这个用户名密码是在conf/users.properties中配置的。
输入用户名密码后便可看到如下图的ActiveMQ掌握台界面了。

4.1.1掌握台先容

Number Of Consumers 消费者 这个是消费者真个消费者数量

Number Of Pending Messages 等待消费的 这个是当前未出行列步队的数量。
可以理解为总吸收数-总出行列步队数

Messages Enqueued 进入行列步队的 进入行列步队的总数量,包括出行列步队的。
这个数量只增不减

Messages Dequeued 出了行列步队的 可以理解为是消费这消费掉的数量

这个要分两种情形理解

在queues里它和进入行列步队的总数量相等(由于一个只会被成功消费一次),如果暂时不等是由于消费者还没来得及消费。

在 topics里 它由于多消费者从而导致数量会比入行列步队数高。

大略的理解上面的意思便是

当有一个进入这个行列步队时，等待消费的是1，进入行列步队的是1。

当消费后，等待消费的是0，进入行列步队的是1，出行列步队的是1.

在来一条时，等待消费的是1，进入行列步队的便是2.

没有消费者时 Pending Messages 和 入行列步队数量一样

有消费者消费的时候 Pedding会减少 出行列步队会增加

到末了 便是 入行列步队和出行列步队的数量一样多

以此类推，进入行列步队的和出行列步队的是池子，等待消费的是水流。

4.2 、实现点对点通讯模式

利用ActiveMQ完成点对点（p2p）通讯模式

引入pom文件依赖

<dependencies><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-core</artifactId><version>5.7.0</version></dependency></dependencies>

生产者

public class Producter {public static void main(String[] args) throws JMSException {// ConnectionFactory ：连接工厂，JMS 用它创建连接ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD, \公众tcp://127.0.0.1:61616\公众);// JMS 客户端到JMS Provider 的连接Connection connection = connectionFactory.createConnection();connection.start();// Session： 一个发送或吸收的线程Session session = connection.createSession(Boolean.falst, Session.AUTO_ACKNOWLEDGE);// Destination ：的目的地;发送给谁.// 获取session把稳参数值my-queue是Query的名字Destination destination = session.createQueue(\"大众my-queue\"大众);// MessageProducer：生产者MessageProducer producer = session.createProducer(destination);// 设置不持久化producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 发送一条for (int i = 1; i <= 5; i++) {sendMsg(session, producer, i);}session.commit();connection.close();}/ 在指定的会话上，通过指定的生产者发出一条 @param session 会话 @param producer 生产者 /public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {// 创建一条文本TextMessage message = session.createTextMessage(\"大众Hello ActiveMQ！
\公众 + i);// 通过生产者发出producer.send(message);}}

消费者

public class JmsReceiver {public static void main(String[] args) throws JMSException {// ConnectionFactory ：连接工厂，JMS 用它创建连接ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD, \公众tcp://127.0.0.1:61616\公众);// JMS 客户端到JMS Provider 的连接Connection connection = connectionFactory.createConnection();connection.start();// Session： 一个发送或吸收的线程Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);// Destination ：的目的地;发送给谁.// 获取session把稳参数值xingbo.xu-queue是一个做事器的queue，须在在ActiveMq的console配置Destination destination = session.createQueue(\"大众my-queue\公众);// 消费者，吸收者MessageConsumer consumer = session.createConsumer(destination);while (true) {TextMessage message = (TextMessage) consumer.receive();if (null != message) {System.out.println(\"大众收到：\"大众 + message.getText());session.commit();} elsebreak;}session.close();connection.close();}}

4.3 、JMS可靠机制

ActiveMQ签收机制：

客戶端成功吸收一条的标志是一条被签收，成功应答。

的签收环境分两种：

1、带事务的session

如果session带有事务，并且事务成功提交，则被自动签收。
如果事务回滚，则会被再次传送。

2、不带事务的session

不带事务的session的签收办法，取决于session的配置。

Activemq支持一下三種模式：

Session.AUTO_ACKNOWLEDGE 自动签收

Session.CLIENT_ACKNOWLEDGE 客戶端调用acknowledge方法手动签收

textMessage.acknowledge();//手动签收

Session.DUPS_OK_ACKNOWLEDGE 不是必须签收，可能会重复发送。
在第二次重新传送的时候，

只有在被确认之后，才认为已经被成功地消费了。
的成功消费常日包含三个阶段：客户吸收、客户处理和被确认。
在事务性会话中，当一个事务被提交的时候，确认自动发生。
在非事务性会话中，何时被确认取决于创建会话时的应答模式（acknowledgement mode）。
该参数有以下三个可选值：

Number Of Consumers 消费者 这个是消费者真个消费者数量

Number Of Pending Messages 等待消费的 这个是当前未出行列步队的数量。
可以理解为总吸收数-总出行列步队数

Messages Enqueued 进入行列步队的 进入行列步队的总数量,包括出行列步队的。
这个数量只增不减

Messages Dequeued 出了行列步队的 可以理解为是消费这消费掉的数量

4.4 、发布订阅

生产者:

public class TOPSend {private static String BROKERURL = \"大众tcp://127.0.0.1:61616\"大众;private static String TOPIC = \公众my-topic\公众;public static void main(String[] args) throws JMSException {start();}static public void start() throws JMSException {System.out.println(\"大众生产者已经启动....\公众);// 创建ActiveMQConnectionFactory 会话工厂ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);Connection connection = activeMQConnectionFactory.createConnection();// 启动JMS 连接connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);MessageProducer producer = session.createProducer(null);producer.setDeliveryMode(DeliveryMode.PERSISTENT);send(producer, session);System.out.println(\"大众发送成功!\"大众);connection.close();}static public void send(MessageProducer producer, Session session) throws JMSException {for (int i = 1; i <= 5; i++) {System.out.println(\公众我是\"大众 + i);TextMessage textMessage = session.createTextMessage(\公众我是\"大众 + i);Destination destination = session.createTopic(TOPIC);producer.send(destination, textMessage);}}}

消费者:

public class TopReceiver {private static String BROKERURL = \"大众tcp://127.0.0.1:61616\公众;private static String TOPIC = \公众my-topic\"大众;public static void main(String[] args) throws JMSException {start();}static public void start() throws JMSException {System.out.println(\公众消费点启动...\公众);// 创建ActiveMQConnectionFactory 会话工厂ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);Connection connection = activeMQConnectionFactory.createConnection();// 启动JMS 连接connection.start();// 不开缘由物，紧张发送消费者,则表示已经签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 创建一个行列步队Topic topic = session.createTopic(TOPIC);MessageConsumer consumer = session.createConsumer(topic);// consumer.setMessageListener(new MsgListener());while (true) {TextMessage textMessage = (TextMessage) consumer.receive();if (textMessage != null) {System.out.println(\"大众接管到:\"大众 + textMessage.getText());// textMessage.acknowledge();// 手动签收// session.commit();} else {break;}}connection.close();}}

4.5 、SpringBoot整合ActiveMQ

生产者:

4.5.1 引入 maven依赖

<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.4.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- spring boot web支持：mvc,aop... --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build>

4.5.2 引入 YML配置

spring: activemq: broker-url: tcp://127.0.0.1:61616 user: admin password: adminqueue: kmx.pas.job.sgtest

4.5.3 创建QueueConfig

@Configurationpublic class QueueConfig {@Value(\"大众${queue}\"大众)private String queue;@Beanpublic Queue logQueue() {return new ActiveMQQueue(queue);}}

4.5.4 创建Producer

@Component@EnableSchedulingpublic class Producer {@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;@Autowiredprivate Queue queue;@Scheduled(fixedDelay = 5000)public void send() {jmsMessagingTemplate.convertAndSend(queue, \"大众测试行列步队\"大众 + System.currentTimeMillis());}}

4.5.5 启动

@SpringBootApplicationpublic class App {public static void main(String[] args) {SpringApplication.run(App.class, args);}}

消费者:

4.5.1 引入 maven依赖

<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.4.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- spring boot web支持：mvc,aop... --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build>

4.5.2 引入 YML配置

application.yml

spring: activemq: broker-url: tcp://127.0.0.1:61616 user: admin password: adminqueue: kmx.pas.job.sgtest

4.5.3 创建Consumer

@Component

public class Consumer {

@JmsListener(destination = \"大众${queue}\"大众)

public void receive(String msg) {

System.out.println(\"大众监听器收到msg:\"大众 + msg);

}

}

4.5.4 启动

@SpringBootApplicationpublic class App {public static void main(String[] args) {SpringApplication.run(App.class, args);}}