你开了个好头. Redis Pub / Sub涉及至少两方：

>订阅者

>和出版商

订阅者(猜测并不奇怪)订阅频道,模式或两者.

发布者将消息发布到频道.此设置也需要反映在您的代码中.

我使用RedisPubSubAdapter扩展了你的代码,因此代码不需要实现所有方法,只需要我们感兴趣的方法,比如message(channel,message)：

RedisClient client = RedisClient.create("redis://" + host + "/0");

StatefulRedisPubSubConnection con = client.connectPubSub();

RedisPubSubListener listener = new RedisPubSubAdapter() {

@Override

public void message(String channel, String message) {

System.out.println(String.format("Channel: %s, Message: %s", channel, message));

}

};

con.addListener(listener);

RedisPubSubCommands sync = con.sync();

sync.subscribe("channel");

一旦添加了监听器并且客户端订阅了一个频道,它就可以接收发布/订阅消息了.当通知到达时,生菜会调用听众的方法.此时,重要的是要了解通知是在I / O线程上处理的,该线程与设置客户端和订阅的线程不同.

让我们来发送方.要向您的频道发送消息,您需要一个额外的连接(或者使用redis-cli并发出PUBLISH频道消息).

StatefulRedisConnection sender = client.connect();

sender.sync().publish("channel", "Message 1");

sender.sync().publish("channel", "Message 2");

Redis将在名为channel的频道上发布消息1和消息1(不是广告素材名称,但它现在可以完成这项工作).

如果你连续执行代码并在发送一条消息后稍等一下,那么你很有可能通知监听器并且你看到一些系统输出如：

Channel: channel, Message: Message 1

Channel: channel, Message: Message 2

异步：有什么影响？

现在是异步性的棘手部分.在某些情况下使用异步通信是有益的,但增加了复杂性如果您可以在结果到达之前完成工作(某些计算直到您需要结果),或者您只想启动I / O并释放您正在处理的线程.服务器应用程序是异步模式的良好环境.典型的服务器具有有限的线程资源,并且它一直运行直到它关闭.在服务器启动时,您将注册订阅.一旦消息进入,它就会在I / O线程上处理,并且会调用您的侦听器

在独立应用程序中使用异步命令执行(比如一个简单的主程序)时,您就有了顺序流程.异步消息传递将导致程序在代码流完成后退出.这并不一定意味着接收或处理了发布/订阅消息.如果你在main中逐个运行两个代码块,你很可能根本看不到任何输出,因为程序终止的速度比I / O可能发生的要快.现在同步发挥作用.如何处理同步有很多可能性,但现在让我们看看两种替代方案：

> CountDownLatch：在发布程序流之前需要执行许多操作

> Thread.sleep(…)：等待几毫秒

CountDownLatch使用

final CountDownLatch latch = new CountDownLatch(2);

RedisPubSubListener listener = new RedisPubSubAdapter() {

@Override

public void message(String channel, String message) {

System.out.println(String.format("Channel: %s, Message: %s", channel, message));

latch.countDown();

}

};

// ...

sender.sync().publish("channel", "Message 2");

latch.await();

在上面的代码中,CountDownLatch被设置为倒计时两次(latch.countDown()).调用latch.await()会阻塞主线程(程序流)并使其等待,直到CountDownLatch倒计时,从而释放程序以继续.

Thread.sleep(…)使用

RedisPubSubListener listener = new RedisPubSubAdapter() {

@Override

public void message(String channel, String message) {

System.out.println(String.format("Channel: %s, Message: %s", channel, message));

}

};

con.addListener(listener);

RedisPubSubCommands sync = con.sync();

sync.subscribe("channel");

StatefulRedisConnection sender = client.connect();

sender.sync().publish("channel", "Message 1");

sender.sync().publish("channel", "Message 2");

latch.await();

Thread.sleep(1000);

此代码使用Thread.sleep(1000);等待一秒钟(在主线程上).这应该足以接收消息.不要这样做.这种方法快速而且很脏,可能适合播放和调试,但在合理的代码中避免使用Thread.sleep.

有两个参与者的事情

Redis对订阅频道/模式的连接施加约束：一旦订阅,您只能执行SUBSCRIBE,PSUBSCRIBE,UNSUBSCRIBE,PUNSUBSCRIBE,PING和QUIT命令.不允许在该连接上执行PUBLISH.因此,您需要使用其他连接.