AMQ: Publisher/Subscriber(发布/订阅者)消息模式开发流程

[b]1


、订阅者（
Subscriber


）开发流程（
TopicListener.java


）：

[/b]

1.1

实现

MessageListener

接口：

在

onMessage()

方法中监听发布者发出的消息队列，并做相应处理。

1.2

创建

Connection

：

根据

url

，

user

和

password

创建一个

jms Connection

。

1.3

创建

Session

：

在

connection

的基础上创建一个

session

，同时设置是否支持事务和

ACKNOWLEDGE

标识。

1.4

创建

Topic

：



创建

2

个

Topic

，

topictest.messages



用于接收发布者发出的消息，

topictest.control

用于向发布者发送消息，实现双方的交互。

1.5

创建

consumer

和

producer

对象：



根据

topictest.messages

创建

consumer

，根据

topictest.control

创建

producer

。

1.6

接收处理消息：



在

onMessage()

方法中，对收到的消息进行处理，可直接简单在本地显示消息，或者根据消息内容不同处理对应的业务逻辑（比如：数据库更新、文件操作等
等），并且可以使用

producer

对象将处理结果返回给发布者。




[b][b]2


、发布者（
Publisher


）开发流程（
TopicPublisher.java


）：

[/b]
[/b]

[b]2.1

[/b]

[b]实
现
[/b]

[b]MessageListener
[/b]

[b]接口：
[/b]

[b]在
[/b]

[b]onMessage()
[/b]

[b]方法中接收订阅者的反馈消息。
[/b]

[b]2.2

[/b]

[b]创
建
[/b]

[b]Connection
[/b]

[b]：
[/b]

[b]根据
[/b]

[b]url
[/b]

[b]，
[/b]

[b]user
[/b]

[b]和
[/b]

[b]password
[/b]

[b]创建一个
[/b]

[b]jms
Connection
[/b]

[b]。
[/b]

[b]2.3

[/b]

[b]创
建
[/b]

[b]Session
[/b]

[b]：
[/b]

[b]在
[/b]

[b]connection
[/b]

[b]的基础上创建一个
[/b]

[b]session
[/b]

[b]，同时设置是否支持事务和
[/b]

[b]ACKNOWLEDGE
[/b]

[b]标识。
[/b]

[b]2.4

[/b]

[b]创
建
[/b]

[b]Topic
[/b]

[b]：
[/b]




[b]创建
[/b]

[b]2
[/b]

[b]个
[/b]

[b]Topic
[/b]

[b]，
[/b]

[b]topictest.messages
[/b]

[b]用于向订阅者发布消息，
[/b]

[b]topictest.control
[/b]

[b]用于接收订阅者反馈的消息。这
[/b]

[b]2
[/b]

[b]个
[/b]

[b]topic
[/b]

[b]与订阅者开发流程中的
[/b]

[b]topic
[/b]

[b]是一一对应的。
[/b]

[b]2.5

[/b]

[b]创
建
[/b]

[b]consumer
[/b]

[b]和
[/b]

[b]producer
[/b]

[b]对象：
[/b]




[b]根据
[/b]

[b]topictest.messages
[/b]

[b]创建
[/b]

[b]publisher
[/b]

[b]；
[/b]

[b]根据
[/b]

[b]topictest.control
[/b]

[b]创建
[/b]

[b]consumer
[/b]

[b]，同时监听订阅者反馈的消息。
[/b]

[b]2.6

[/b]

[b]给
所有订阅者发送消息，并接收反馈消息：
[/b]




[b]示例代码中，一共重复
[/b]

[b]10
[/b]

[b]轮操作。
[/b]

[b]每轮先向所有订阅者发送
[/b]

[b]2000
[/b]

[b]个消息；
[/b]

[b]然后堵塞线程，开始等待；
[/b]

[b]最后通过
[/b]

[b]onMessage()
[/b]

[b]方法，接收到订阅者反馈的“
[/b]

[b]REPORT
[/b]

[b]”类信息后，才
[/b]

[b]print
[/b]

[b]反馈信息并解除线程堵塞，进入下一轮。
[/b]

[b][b]注：

可
同时运行多个订阅者测试查看此模式效果
[/b]
[/b]