文档章节

ActiveMQ发布订阅模式

postdep
 postdep
发布于 2015/08/24 22:14
字数 524
阅读 124
收藏 1

ActiveMQ的另一种模式就SUB/HUB即发布订阅模式,是SUB/hub就是一拖N的USB分线器的意思。意思就是一个来源分到N个出口。还是上节的例子,当一个订单产生后,后台N个系统需要联动,但有一个前提是都需要收到订单信息,那么我们就需要将一个生产者的消息发布到N个消费者。

生产者:

 

复制代码
try { //Create the Connection Factory  IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/"); using (IConnection connection = factory.CreateConnection())
                { //Create the Session  using (ISession session = connection.CreateSession())
                    { //Create the Producer for the topic/queue  IMessageProducer prod = session.CreateProducer( new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing")); //Send Messages  int i = 0; while (!Console.KeyAvailable)
                        {
                            ITextMessage msg = prod.CreateTextMessage();
                            msg.Text = i.ToString();
                            Console.WriteLine("Sending: " + i.ToString());
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);

                            System.Threading.Thread.Sleep(5000);
                            i++;
                        }
                    }
                }

                Console.ReadLine();
            } catch (System.Exception e)
            {
                Console.WriteLine("{0}", e.Message);
                Console.ReadLine();
            }
复制代码

假设生产者每5秒发送一次消息:

wps3E59.tmp

消费者:

复制代码
static void Main(string[] args)
        { try { //Create the Connection factory  IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/"); //Create the connection  using (IConnection connection = factory.CreateConnection())  
                {  
                    connection.ClientId = "testing listener1";  
                    connection.Start(); //Create the Session  using (ISession session = connection.CreateSession())  
                    { //Create the Consumer  IMessageConsumer consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"), "testing listener1", null, false);  
                          
                        consumer.Listener += new MessageListener(consumer_Listener);  
  
                        Console.ReadLine();  
                    }  
                    connection.Stop();  
                    connection.Close();  
                }  
            } catch (System.Exception e)  
            {  
                Console.WriteLine(e.Message);  
            }  
        } static void consumer_Listener(IMessage message)  
        { try {  
                ITextMessage msg = (ITextMessage)message;  
                Console.WriteLine("Receive: " + msg.Text);  
           } catch (System.Exception e)  
            {  
                Console.WriteLine(e.Message);  
            }  
        }
复制代码

 

启动一个消费者:

wps3E5A.tmp

我们发现他是从15开始的,而不是像上节一样从头开始,再启动另一个消费者:

wps3E5B.tmp

我们发现就是从启动时开始接受消息的,之前的消息就丢失了。

整体状态如下:

wps3E6B.tmp

我们观察管理界面:

wps3E6C.tmp

产生了一个testing的Topics,而订阅方有2个都订阅的是testing:

wps3E6D.tmp

这样只需要在需要获取消息的地方订阅即可及时获得。

源代码下载

本文转载自:http://www.cnblogs.com/madyina/p/4127144.html

共有 人打赏支持
postdep

postdep

粉丝 81
博文 255
码字总数 261672
作品 0
武汉
高级程序员
消息队列篇—详谈ActiveMQ消息队列模式的分析及使用

消息队列(Message Queue)是分布式系统中重要的组件,通用使用场景可以简单地描述为当不需要立即获得结果,但是并发量需控制时就需要使用消息队列。消息列队有两种消息模式,一种是点对点的...

afreon
04/22
0
0
Linux 安装ActiveMQ(使用Mac远程访问)

阅读本文需要安装JDK 一 ActiveMQ简介 activemq是用java语言编写的一款开源消息总线 activemq是apache出品 activemq消息的传递有两种类型 一种是点对点(即一个生产者和一个消费者一一对应) 另...

梦三
07/15
0
0
深入浅出 消息队列 ActiveMQ

一、 概述与介绍 ActiveMQ 是Apache出品,最流行的、功能强大的即时通讯和集成模式的开源服务器。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。提供客户端支持跨语言和...

亮liang
2015/05/27
0
0
SpringBoot JMS(ActiveMQ) 使用实践

ActiveMQ 1. 下载windows办的activeMQ后,在以下目录可以启动: 2. 启动后会有以下提示 3. 所以我们可以通过http://localhost:8161访问管理页面,通过tcp://localhost:61616来连接消息服务器...

yawnSilence
2017/10/28
0
12
ActiveMQ专题2: 持久化

AMQ的持久化问题 前言 前面一篇AMQ专题中,我们发现对于Topic这种类型的消息,即使将deliveryMode设置为持久化,只要生产者在消费者之前启动。消息生产者发布的消息还是会丢失。这是符合JMS...

槟城码农
08/30
0
0

没有更多内容

加载失败,请刷新页面

加载更多

what f,,

anlve
39分钟前
0
0
初级开发-编程题

` public static void main(String[] args) { System.out.println(changeStrToUpperCase("user_name_abc")); System.out.println(changeStrToLowerCase(changeStrToUpperCase("user_name_abc......

小池仔
今天
9
0
现场看路演了!

HiBlock
昨天
19
0
Rabbit MQ基本概念介绍

RabbitMQ介绍 • RabbitMQ是一个消息中间件,是一个很好用的消息队列框架。 • ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的s...

寰宇01
昨天
13
0
官方精简版Windows10:微软自己都看不过去了

微软宣布,该公司正在寻求解决方案,以减轻企业客户的Windows 10规模。该公司声称,企业客户下载整个Windows 10文件以更新设备既费钱又费时。 微软宣布,该公司正在寻求解决方案,以减轻企业...

linux-tao
昨天
20
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部