使用Hazelcast发布和订阅

几周前，我写了一篇有关Hazelcast入门的博客，描述了创建分布式地图，列表和队列是多么简单。当时我提到Hazelcast还能做很多其他事情。该博客快速浏览了Hazelcast的另一个功能：基于Publish / Subscribe模式的广播消息系统。这采用通常的格式，即邮件发件人应用通过该格式发布有关特定主题的邮件。这些消息并不针对任何特定的客户端，而是可以由对主题感兴趣的任何客户端读取。

发布和订阅的明显场景来自高金融和做市商的世界。做市商买卖股票之类的金融工具，并通过在通常是电子的市场上宣传买卖价格来竞争业务。为了使用Hazelcast实现非常简单的做市商方案，我们需要三个类： StockPrice bean， MarketMaker和Client 。

以下代码已添加到我在Github上可用的现有Hazelcast项目中。无需担心其他POM依赖项。

public class StockPrice implements Serializable { private static final long serialVersionUID = 1L; private final BigDecimal bid; private final BigDecimal ask; private final String code; private final String description; private final long timestamp; /** * Create a StockPrice for the given stock at a given moment */ public StockPrice(BigDecimal bid, BigDecimal ask, String code, String description, long timestamp) { super(); this.bid = bid; this.ask = ask; this.code = code; this.description = description; this.timestamp = timestamp; } public BigDecimal getBid() { return bid; } public BigDecimal getAsk() { return ask; } public String getCode() { return code; } public String getDescription() { return description; } public long getTimestamp() { return timestamp; } @Override public String toString() { StringBuilder sb = new StringBuilder("Stock - "); sb.append(code); sb.append(" - "); sb.append(description); sb.append(" - "); sb.append(description); sb.append(" - Bid: "); sb.append(bid); sb.append(" - Ask: "); sb.append(ask); sb.append(" - "); SimpleDateFormat df = new SimpleDateFormat("HH:MM:SS"); sb.append(df.format(new Date(timestamp))); return sb.toString(); } 
}

StockPrice bean具有所有常用的获取器和设置器，可以在任何给定时间模拟股票的买价和买价（以正常语言进行买卖），并且MarketMaker类使用Hazelcast发布这些bean。

通常，做市商会在一种以上的金融工具中发布价格；但是，为简单起见， MarketMaker在此演示中仅发布单个价格。

public class MarketMaker implements Runnable { private static Random random = new Random(); private final String stockCode; private final String description; private final ITopic<StockPrice> topic; private volatile boolean running; public MarketMaker(String topicName, String stockCode, String description) { this.stockCode = stockCode; this.description = description; this.topic = createTopic(topicName); running = true; } @VisibleForTesting ITopic<StockPrice> createTopic(String topicName) { HazelcastInstance hzInstance = Hazelcast.newHazelcastInstance(); return hzInstance.getTopic(topicName); } public void publishPrices() { Thread thread = new Thread(this); thread.start(); } @Override public void run() { do { publish(); sleep(); } while (running); } private void publish() { StockPrice price = createStockPrice(); System.out.println(price.toString()); topic.publish(price); } @VisibleForTesting StockPrice createStockPrice() { double price = createPrice(); DecimalFormat df = new DecimalFormat("#.##"); BigDecimal bid = new BigDecimal(df.format(price - variance(price))); BigDecimal ask = new BigDecimal(df.format(price + variance(price))); StockPrice stockPrice = new StockPrice(bid, ask, stockCode, description, System.currentTimeMillis()); return stockPrice; } private double createPrice() { int val = random.nextInt(2010 - 1520) + 1520; double retVal = (double) val / 100; return retVal; } private double variance(double price) { return (price * 0.01); } private void sleep() { try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } } public void stop() { running = false; } public static void main(String[] args) throws InterruptedException { MarketMaker bt = new MarketMaker("STOCKS", "BT.L", "British Telecom"); MarketMaker cbry = new MarketMaker("STOCKS", "CBRY.L", "Cadburys"); MarketMaker bp = new MarketMaker("STOCKS", "BP.L", "British Petrolium"); bt.publishPrices(); cbry.publishPrices(); bp.publishPrices(); } }

像往常一样，设置Hazelcast相当简单，上面MarketMaker类中的大多数代码与Hazelcast无关。该课程分为两部分：建筑价格和出版价格。构造函数采用三个参数，将其存储起来以备后用。它还创建一个Hazelcast实例，并通过私有createTopic()方法注册一个名为"STOCKS"的简单主题。如您所料，创建Hazelcast实例并注册主题需要两行代码，如下所示：

ITopic<StockPrice> createTopic(String topicName) { HazelcastInstance hzInstance = Hazelcast.newHazelcastInstance(); return hzInstance.getTopic(topicName); }

该类的其余部分使用线程来调用MarketMaker的run()方法来运行价格发布机制。此方法生成随机出价，为关联的股票代码要价，并使用Hazelcast发布。使用以下单行代码即可完成发布：

topic.publish(price);

MarketMaker类的最后一部分是main()方法，其作用是创建多个MarketMaker实例并使它们运行。

现在，Hazelcast知道了我们不断变化的股价，接下来要做的就是整理客户代码。

public class Client implements MessageListener<StockPrice> { public Client(String topicName) { HazelcastInstance hzInstance = Hazelcast.newHazelcastInstance(); ITopic<StockPrice> topic = hzInstance.getTopic(topicName); topic.addMessageListener(this); } /** * @see com.hazelcast.core.MessageListener#onMessage(com.hazelcast.core.Message) */ @Override public void onMessage(Message<StockPrice> arg0) { System.out.println("Received: " + arg0.getMessageObject().toString()); } public static void main(String[] args) { new Client("STOCKS"); } }

与任何消息传递系统一样，消息发送者代码必须知道呼叫谁和呼叫什么。客户端通过创建Hazelcast实例并在"STOCKS"主题中注册兴趣来实现“调用什么"STOCKS" ，方法与发布者相同，如下所示：

HazelcastInstance hzInstance = Hazelcast.newHazelcastInstance(); ITopic<StockPrice> topic = hzInstance.getTopic(topicName); topic.addMessageListener(this);

客户端实现Hazelcast的MessageListener接口及其单一方法onMessage()实现“呼叫”

@Override public void onMessage(Message<StockPrice> arg0) { System.out.println("Received: " + arg0.getMessageObject().toString()); }

客户端代码的最后一部分是其main()方法，该方法创建一个客户端实例。

最后要做的是运行代码。为此，我仅将所有必需的JAR文件放在一个目录中，只需考虑两个：hazel cast-3.1.jar和guava-13.0.1.jar。

屏幕截图2013年12月7日的10.51.43

完成后，我转到项目的classes目录：

cd /Users/Roger/git/captaindebug/hazelcast/target/classes

…并解雇了发布者

java -cp ./:/Users/Roger/tmp/mm/guava-13.0.1.jar:/Users/Roger/tmp/mm/hazelcast-3.1.jar com.captaindebug.hazelcast.pubsub.MarketMaker

……然后是客户。

java -cp ./:/Users/Roger/tmp/mm/guava-13.0.1.jar:/Users/Roger/tmp/mm/hazelcast-3.1.jar com.captaindebug.hazelcast.pubsub.Client

当然，如果您正在使用此粗略且已准备好的技术在计算机上运行此程序，则请记住将其替换
/Users/Roger/tmp/mm以及放置这些JAR文件副本的路径。

如果您在一个终端中运行MarketMaker发布者，并在其他两个终端中运行几个客户，那么您将得到类似的信息，在这里您可以看到正在发布的价格以及客户正在接收更新。

屏幕截图2013年12月22日的17.40.07

关于Hazelcast的一件事要注意的是，“ 集群 ”是指Hazelcast实例的集群，而不是JVM的集群。在您为每个应用程序请求多个Hazelcast实例之前，这是不明显的。当其他客户端加入集群时，您将看到类似以下内容：

Members [5] {
Member [192.168.0.7]:5701
Member [192.168.0.7]:5702
Member [192.168.0.7]:5703
Member [192.168.0.7]:5704 this
Member [192.168.0.7]:5705
}

在上面的日志中，有两个侦听器条目，每个侦听器条目一个，每个发布者条目三个，在MarketMaker的main()方法中启动的每个MarketMaker实例一个。

屏幕快照2013-12-07 at 11.16.21
这里要考虑的事情是，是否每个对象实例创建一个Hazelcast实例是一种好习惯（就像我在示例代码中所做的那样），还是在代码中有一个static Hazelcast实例更好。我不确定该答案是什么，因此，如果有任何Hazelcast专家正在阅读此书，请告诉我。