Lettuce Pub/Sub

Redis Developer Course	Redis Technical Support	Redis Enterprise Server

Pub/Sub

Lettuce-6.1.7, Redis-7.0.2 Example
Lettuce-4.4.2, Redis-4.0.6 Example

Lettuce-6.1.7, Redis-7.0.2 Example

LettucePubsubCmd.java
/*------------------------------------------------------- * 시작하면 ch01로 subscribe한다. * 1번 publish를 선택하면 ch01로 Hello 메시지를 보낸다. * cli에서 ch01로 subscribe해서 받아보든지, ch01로 publish해본다. */ import java.util.Scanner; import io.lettuce.core.pubsub.RedisPubSubAdapter; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands; public class LettucePubsubCmd { Scanner scanner = new Scanner(System.in); Util util = new Util(); int idx = 1; String channel = "ch01"; public LettucePubsubCmd() { // Subscribe channel RedisPubSubAdapter<String, String> listener = new RedisPubSubAdapter<String, String>() { @Override public void message(String channel, String message) { System.out.println(String.format("subscribe -> Channel: %s, Message: %s", channel, message)); } }; StatefulRedisPubSubConnection<String, String> pubsubConn = Main.redisClient.connectPubSub(); pubsubConn.addListener(listener); RedisPubSubAsyncCommands<String, String> async = pubsubConn.async(); async.subscribe(channel); while(true) { System.out.println("1:PUBLISH\n"); System.out.println("선택(exit:99)> "); String input = scanner.nextLine(); int i = util.checkInt(input); switch (i) { case 1: publish(); break; case 99: break; default: System.out.println("잘못 선택했습니다. 다시 선택하세요. "); } if (i==99) break; } } // PUBLISH channel message public void publish() { try { Main.syncCmd.publish(channel, "Hello"+idx); System.out.println("publish "+channel+" Hello"+idx); idx++; } catch (Exception e) { System.out.println(e); } } }

Lettuce-4.4.2, Redis-4.0.6 Example

Class 변수로 선언

String channel = "ch01";
long idx = 1;

PUBLISH

// PUBLISH channel message
protected void publish() {
	try {
		Redis.syncCmd.publish(channel, "Hello"+idx);
		System.out.println("publish "+channel+" Hello"+idx);
		idx++;
	} catch (Exception e) {
		System.out.println(e);
	}
}

// Subscribe channel 
RedisPubSubAdapter<String, String> listener = new RedisPubSubAdapter<String, String>() {
		@Override
		public void message(String channel, String message) {
				System.out.println(String.format("subscribe -> Channel: %s, Message: %s", channel, message));
		}
};
StatefulRedisPubSubConnection<String, String> pubsubConn = Redis.redisClient.connectPubSub();
pubsubConn.addListener(listener);
RedisPubSubAsyncCommands<String, String> async = pubsubConn.async();
async.subscribe(channel);

Subscribe for Sentinel failover

마스터 서버 다운이 발생하면 센티널은 Failover를 진행한다. Failover를 진행하면서 각 단계마다 로그에 기록하고 단계명을 channel명으로 해서 관련 정보를 publish한다. Client는 "+switch-master" 메시지를 받으면 기존 connection을 버리고 새 마스터에 connect한다. 아래 예제에서는 "+sdown", "+odown"도 받게했다. Sentinel 관련 자세한 내용은 여기를 보세요.

// Subscribe: Sentinel -> check failover
RedisPubSubAdapter<String, String> listener2 = new RedisPubSubAdapter<String, String>() {
		@Override
		public void message(String channel, String message) {
				System.out.println(String.format("subscribe -> Channel: %s, Message: %s", channel, message));
				if (channel.equals("+switch-master")) {
					String items[] = message.split(" ");
					int len = items.length;
					String ip = items[len-2];
					int port = Integer.parseInt(items[len-1]);
					System.out.println(ip+":"+port);
					Redis.conn2(ip, port);
				}
		}
};
StatefulRedisPubSubConnection<String, String> pubsubConn2 = Redis.sentinelClient.connectPubSub();
pubsubConn2.addListener(listener2);
RedisPubSubAsyncCommands<String, String> async2 = pubsubConn2.async();
async2.subscribe("+sdown");
async2.subscribe("+odown");
async2.subscribe("+switch-master");

<< Hashes	Pub/Sub	Common Keys >>

Email 답글이 올라오면 이메일로 알려드리겠습니다.