博客分类：
订阅发布者模式本质上也是一种生产者消费者模式，订阅者是消费者，发布者是生产者。如果一定要说个区别，就是抽象级别的区别吧。
订阅者肯定是个消费者，但消费者不一定是订阅者，发布者一定是个生产者，但生产者不一定是个发布者。
订阅发布者模式有时也称为观察者模式，订阅发布者（观察这和被观察者）存在着主动被动的关系，而生产者消费者比较中性吧。 订阅发布模式定义了一种一对多的依赖关系，让多个订阅者对象同时监听某一个主题对象。这个主题对象在自身状态变化时，会通知所有订阅者对象，使它们能够自动更新自己的状态。而生产者消费者关系可以是1对1，1对多，多对1，多对多关系
补充：在23种设计模式中的观察者模式中，并没有中间介-队列的概念，但生产者消费者模式再多线程环境下好像天生就有队列的概念。在订阅发布者之间引入消息队列后，可以实现订阅者和发布者之间的解耦，任务可以很好的以异步方式进行处理，所以说是否有中间队列不是订阅发布者模式和生产者消费者模式的区别
如下蓝色字体部分为参考内容：
如下摘自：/article/javascript-observer-pattern.html
那么到底什么是观察者模式呢. 先看看生活中的观察者模式。
好莱坞有句名言. “不要给我打电话， 我会给你打电话”. 这句话就解释了一个观察者模式的来龙去脉。 其中“我”是发布者， “你”是订阅者。
再举个例子，我来公司面试的时候，完事之后每个面试官都会对我说：“请留下你的联系方式， 有消息我们会通知你”。 在这里“我”是订阅者， 面试官是发布者。所以我不用每天或者每小时都去询问面试结果， 通讯的主动权掌握在了面试官手上。而我只需要提供一个联系方式。
zhouchaofei2010
浏览: 640136 次
来自: 上海
Saro 写道在log4j配置里把mapper所在包设为deb ...
在log4j配置里把mapper所在包设为debug就行了，参 ...
杀手请杀人 写道能看到scala太难得了
能看到scala太难得了
(window.slotbydup=window.slotbydup || []).push({
id: '4773203',
container: s,
size: '200,200',
display: 'inlay-fix'redis使用（三）：事务，过期时间，排序，订阅/发布，持久化
或者 DISCARD (取消)
127.0.0.1:6379&
127.0.0.1:6379&
127.0.0.1:6379&
127.0.0.1:6379&
redis中事务是一组命令集合，EXEC时会执行这组命令，这组命令执行过程中不会执行其它命令，
1)语法错误, 如：命令不存在，命令参数个数不对
如果有语法错误,这组命令中的所有命令都不会执行;
2)运行错误,命令执行时出现的错误，如果对hash类型的key执行lpush命令;
watch命令，监视一个变量，直到下一个exec,discard,unwatch命令,如果exec执行时监视的键值有变化，则事务不会执行, watch命令可以用来对键值加锁（乐观锁）。
二 .键值的过期时间
设置键的过期时间,秒后会删除键
查看键多长时间后过期，
设置键为永久
127.0.0.1:6379&
127.0.0.1:6379&
127.0.0.1:6379&
127.0.0.1:6379&
127.0.0.1:6379&
#sort 可以对list，set，sorted set, (不支持hash类型)排序，默认转换成浮点数排序，加ALPHA则按字母序排序
127.0.0.1:6379& lrange
letters 0 -1
127.0.0.1:6379& sort letters ALPHA
更多，可以结合BY GET STORE使用
四 .订阅/发布
#publish channelname message 在channelname发布一个消息,此时订阅此channel的客户端会收到消息
#subscribe channelname 订阅channelname的消息
127.0.0.1:6379& subscribe tv1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
3) (integer) 1
1) "message"
#psubscribe globmatch
使用glob通配符订阅
#punsubscribe globmatch
需要与psubscribe成对出现才有用
疑问：订阅状态下，unsubscribe命令不起作用,为什么？？？
127.0.0.1:6379& subscribe tv1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
3) (integer) 1
1) "message"
3) "world"
unsubscribe tv1
Redis支持持久化存储数据，即保存数据到硬盘。支持两种方式持久化，RDB方式和AOF方式;
RDB,即快照，把内存中的数据生成一份副本保存在硬盘上;如下几种情况会执行快照：
1)根据配置规则，如save 100 10 ,表示如果100s内，有10个键值被修改，就执行快照；
2)用户执行save或者bgsave命令，save命令执行期间不会相应客户端的请求，而bgsave可以立即返回，bgsave通过fork出一个子进程来执行快照，主进程继续接收客户端命令;
3)执行flushall（清空所有数据）命令，如果配置了1)中的快照条件，执行flushall时会触发执行快照；
4)主从复制时，从节点向主节点发sync命令后，主节点会执行快照，并开始缓存此后执行的命令，然后把快照和缓存的命令传输给从节点，从而实现从数据库的复制初始化。
RDB的缺点:redis一旦异常退出，只能恢复到最后一次执行快照时的数据。
AOF方式，通过将执行的没条命令写入磁盘来实现持久化,这个过程会降低Redis的性能,AOF文件的存放位置都是dir配置的路径，默认没有开启AOF方式持久化,可以通过配置appendonly
看过本文的人也看了：
我要留言技术领域：
取消收藏确定要取消收藏吗？
删除图谱提示你保存在该图谱下的知识内容也会被删除，建议你先将内容移到其他图谱中。你确定要删除知识图谱及其内容吗？
删除节点提示无法删除该知识节点，因该节点下仍保存有相关知识内容！
删除节点提示你确定要删除该知识节点吗？上一期介绍了我们项目要用到activeMQ来作为jms总线，并且给大家介绍了activeMQ的集群和高可用部署方案，本期给大家再介绍下，如何根据自己的项目需求，更好地使用activeMQ的两种消息处理模式。
queue与topic的技术特点对比
Publish Subscribe messaging 发布订阅消息
Point-to-Point 点对点
topic数据默认不落地，是无状态的。
Queue数据默认会在mq服务器上以文件形式保存，比如Active MQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存储。
完整性保障
并不保证publisher发布的每条数据，Subscriber都能接受到。
Queue保证每条数据都能被receiver接收。
消息是否会丢失
一般来说publisher发布消息到某一个topic时，只有正在监听该topic地址的sub能够接收到消息；如果没有sub在监听，该topic就丢失了。
Sender发送消息到目标Queue，receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取，也不会丢失。
消息发布接收策略
一对多的消息发布接收策略，监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器
一对一的消息发布接收策略，一个sender发送的消息，只能有一个receiver接收。receiver接收完后，通知mq服务器已接收，mq服务器对queue里的消息采取删除或其他操作。
Topic和queue的最大区别在于topic是以广播的形式，通知所有在线监听的客户端有新的消息，没有监听的客户端将收不到消息；而queue则是以点对点的形式通知多个处于监听状态的客户端中的一个。
topic和queue方式的消息处理效率比较
通过增加监听客户端的并发数来验证，topic的消息推送，是否会因为监听客户端的并发上升而出现明显的下降，测试环境的服务器为ci环境的ActiveMQ，客户端为我的本机。
从实测的结果来看，topic方式发送的消息，发送和接收的效率，在一个订阅者和100个订阅者的前提下没有明显差异，但在500个订阅者（线程）并发的前提下，效率差异很明显（由于500线程并发的情况下，我本机的cpu占用率已高达70-90%，所以无法确认是我本机测试造成的性能瓶颈还是topic消息发送方式存在性能瓶颈，造成效率下降如此明显）。
Topic方式发送的消息与queue方式发送的消息，发送和接收的效率，在一个订阅者和100个订阅者的前提下没有明显差异，但在500个订阅者并发的前提下，topic方式的效率明显低于queue。
Queue方式发送的消息，在一个订阅者、100个订阅者和500个订阅者的前提下，发送和接收的效率没有明显变化。
Topic实测数据：
发送者发送的消息总数
所有订阅者接收到消息的总数
消息发送和接收平均耗时
Queue实测数据：
发送者发送的消息总数
所有订阅者接收到消息的总数
消息发送和接收平均耗时
topic方式的消息处理示例
通过客户端代码调用来发送一个topic的消息：
import javax.jms.C
import javax.jms.ConnectionF
import javax.jms.DeliveryM
import javax.jms.D
import javax.jms.MessageP
import javax.jms.S
import javax.jms.TextM
import org.apache.activemq.ActiveMQC
import org.apache.activemq.ActiveMQConnectionF
publicclass SendTopic {
privatestaticfinalint
SEND_NUMBER = 5;
publicstaticvoid sendMessage(Session session, MessageProducer producer)
throws Exception {
int i = 1; i &=
SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage(&ActiveMq发送的消息& + i);
//发送消息到目的地方
out.println(&发送消息：& + &ActiveMq 发送的消息& + i);
producer.send(message);
publicstaticvoid main(String[] args) {
// ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactory connectionF
// Connection：JMS客户端到JMS Provider的连接
Connection connection =
// Session：一个发送或接收消息的线程
// Destination：消息的目的地;消息发送给谁.
// MessageProducer：消息发送者
//构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
connectionFactory =
new ActiveMQConnectionFactory(
ActiveMQConnection.
DEFAULT_USER,
ActiveMQConnection.
DEFAULT_PASSWORD,
&tcp://10.20.8.198:61616&);
//构造从工厂得到连接对象
connection = connectionFactory.createConnection();
connection.start();
//获取操作连接
session = connection.createSession(
true, Session.
AUTO_ACKNOWLEDGE);
//获取session注意参数值FirstTopic是一个服务器的topic(与queue消息的发送相比，这里是唯一的不同)
destination = session.createTopic(&FirstTopic&);
//得到消息生成者【发送者】
producer = session.createProducer(destination);
//设置不持久化，此处学习，实际根据项目决定
producer.setDeliveryMode(DeliveryMode.
PERSISTENT);
//构造消息，此处写死，项目就是参数，或者方法获取
sendMessage(session, producer);
catch (Exception e) {
e.printStackTrace();
null != connection)
connection.close();
catch (Throwable ignore) {
启动多个客户端监听来接收topic的消息：
publicclass ReceiveTopic
implements Runnable {
private StringthreadN
ReceiveTopic(String threadName) {
this.threadName = threadN
publicvoid run() {
// ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactory connectionF
// Connection：JMS客户端到JMS Provider的连接
Connection connection =
// Session：一个发送或接收消息的线程
// Destination：消息的目的地;消息发送给谁.
//消费者，消息接收者
connectionFactory =
new ActiveMQConnectionFactory(
ActiveMQConnection.
DEFAULT_USER,
ActiveMQConnection.
DEFAULT_PASSWORD,&tcp://10.20.8.198:61616&);
//构造从工厂得到连接对象
connection = connectionFactory.createConnection();
connection.start();
//获取操作连接,默认自动向服务器发送接收成功的响应
session = connection.createSession(
false, Session.
AUTO_ACKNOWLEDGE);
//获取session注意参数值FirstTopic是一个服务器的topic
destination = session.createTopic(&FirstTopic&);
consumer = session.createConsumer(destination);
//设置接收者接收消息的时间，为了便于测试，这里设定为100s
TextMessage message = (TextMessage) consumer
.receive(100 * 1000);
null != message) {
out.println(&线程&+threadName+&收到消息:& + message.getText());
catch (Exception e) {
e.printStackTrace();
null != connection)
connection.close();
catch (Throwable ignore) {
publicstaticvoid main(String[] args) {
//这里启动3个线程来监听FirstTopic的消息，与queue的方式不一样三个线程都能收到同样的消息
ReceiveTopic receive1=
new ReceiveTopic(&thread1&);
ReceiveTopic receive2=
new ReceiveTopic(&thread2&);
ReceiveTopic receive3=
new ReceiveTopic(&thread3&);
Thread thread1=
new Thread(receive1);
Thread thread2=
new Thread(receive2);
Thread thread3=
new Thread(receive3);
thread1.start();
thread2.start();
thread3.start();
queue方式的消息处理示例
参考上一期文章：开源jms服务ActiveMQ的负载均衡+高可用部署方案探索。
相关 [activemq queue topic] 推荐：
- Java - 编程语言 - ITeye博客
上一期介绍了我们项目要用到activeMQ来作为jms总线，并且给大家介绍了activeMQ的集群和高可用部署方案，本期给大家再介绍下，如何根据自己的项目需求，更好地使用activeMQ的两种消息处理模式. 1
queue与topic的技术特点对比. Publish Subscribe messaging 发布订阅消息.
- 博客园_首页
摘要：ActiveMQ优化 客户端优化 预取限制. 原文：
/docs/broker/5.4/tuning/GenTuning-Consumer-Prefetch.html. Overview：图列4.1阐明了Broker在等待之前发送给客户端消息的反馈的行为.
- 博客园_首页
/docs/broker/5.4/tuning/PersTuning-SerialToDisk.html.
KahaDB message store：KahaDB 是ActiveMQ Broker 为了高性能而推荐使用的消息存储机制.
- CSDN博客互联网推荐文章
使用目的：将本地产生的消息转发到远程，通过远程服务器来处理消息，处理完成后，再启动消费者处理本地服务器消息（验证消息是否被转走，本地无消息可处理为正常）. 消息在下面的地址被消费，无需任何特别配置，采用默认的配置即可. 生产消息地址为localhost:7001，需要做如下配置. 注意： 表示只有这个队列的会进行桥接转发.
- CSDN博客推荐文章
前面的讨论提到. PageRank忽略了主题相关性，导致结果的. 相关性和主题性降低，对于不同的用户，甚至有很大的差别. 例如，当搜索“苹果”时，一个数码爱好者可能是想要看
iphone 的信息，一个果农可能是想看苹果的价格走势和种植技巧，而一个小朋友可能在找苹果的简笔画. 理想情况下，应该为每个用户维护一套专用向量，但面对海量用户这种方法显然不可行.
- CSDN博客架构设计推荐文章
Activemq是众多开源消息中间件的一种，支持集群，同等网络，自动检测，TCP，SSL，广播，持久化，和J2EE1.4容器无缝结合. 它是apache基金会的一个项目，而且经过多年发展，有了很高的稳定性. 目前被很多知名项目使用，比如Apache serviceMix、FuseESB.
消息中间件一般被用在异步消息通信、整合多个系统的场景，比如你注册CSDN论坛，你填写完注册信息点提交时，它会发一份验证邮箱的验证邮件给到你，这封邮件就可以通过消息中间异步发送给你.
- 博客园_首页
ActiveMQ 是Apache出品， 是最流行和最强大的开源消息总线. 同时完全支持 JMS 1.1和J2EE 1.4规范. 支持多种编程语言和协议编写客户端. 在JMS客户端和消息代理完全支持企业集成模式. 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务). 对Spring的支持, ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性.
消息生产者使用持久（persistent）传递模式发送消息的时候，Producer.send() 方法会被阻塞，直到 broker 发送一个确认消息给生产者，这个确认消息暗示生产者 broker 已经成功地将它发送的消息路由到目标目的并把消息保存到二级存储中. 但有一个例外，当发送方法在一个事物上下文中时，被阻塞的是commit 方法而不是 send 方法.
- CSDN博客架构设计推荐文章
消息持久性对于可靠消息传递来说应该是一种比较好的方法，有了消息持久化，即使发送者和接受者不是同时在线或者消息中心在发送者发送消息后宕机了，在消息中心重新启动后仍然可以将消息发送出去，如果把这种持久化和ReliableMessaging结合起来应该是很好的保证了消息的可靠传送. 消息持久性的原理很简单，就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等，然后试图将消息发送给接收者，发送成功则将消息从存储中删除，失败则继续尝试.
- 企业架构 - ITeye博客
目前常用的消息队列组建无非就是MSMQ和ActiveMQ，至于他们的异同，这里不想做过多的比较. 简单来说，MSMQ内置于微软操作系统之中，在部署上包含一个隐性条件：Server需要是微软操作系统. （对于这点我并去调研过MSMQ是否可以部署在非微软系统，比如:Linux，只是拍脑袋想了想，感觉上是不可以）.
--> 坚持分享优质有趣的原创文章，并保留作者信息和版权声明，任何问题请联系：@。}