redis clients lettuce2 lettuce2_pubsub

Lettuce Pub/Sub

Redis Developer Course	Redis Technical Support	Redis Enterprise Server

Lettuce Pub/Sub

• Publish connection과 Subscribe connection을 구분해서 사용합니다.
  같은 connection에서 subscribe를 사용하는 중에 publish를 실행하면 아래과 같은 Runtime Error가 발생합니다.
  
  Command PUBLISH not allowed while subscribed.
  Allowed commands are: [PSUBSCRIBE, QUIT, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PING]]
• 일반 연결(StatefulRedisConnection)로 publish 명령을 사용할 수 있으나, Publish connection(StatefulRedisPubSubConnection)으로 사용할 것을 권장합니다.
• Redis Standalone 일 경우 Application 실행 중 서버가 다운되었다, 다시 시작하면 Lettuce에서 자동으로 subscribe 명령을 다시 실행해서 등록합니다.
  
• Master/Replica 구성에서 Subscribe(등록)는 Failover 시 새 마스터에 자동으로 전파(등록)되지 않습니다. 새로 Subscribe(등록)해야 합니다.
• Sentinel 구성에서 Failover 시 Lettuce가 자동으로 새 마스터에 Subscribe(등록)를 실행합니다.

사용(실행) 순서

1. http://localhost:8080/pubsubAddListener
   SubConnection -> 리스너(listener)에 RedisPubSubAdapter()을 등록합니다.
   redisSubConn.addListener(new RedisPubSubAdapter<String, String>() { ... }
2. http://localhost:8080/subscribe/ch01
   SubConnection -> Subscribe 명령(메서드)를 실행합니다.
   redisSubCommands.subscribe(channel);
   [주의] subscribe()를 while(true) 안에 두지 마세요.
   레디스에서 subscribe는 등록하는 것으로 한 번만 실행하면 됩니다.
   소스 코드에 해당 설명이 조금 더 자세히 있습니다.
3. http://localhost:8080/publish/ch01:Hello
   PubConnection -> Publish 명령(메서드)를 실행합니다.
   redisCommands.publish(channel, message);

1. Java Publish/Subscribe Source

Java Lettuce를 사용한 Publish/Subscribe 명령 사용법입니다.

Publish/Subscribe Source

Redis10_PubSub.java
package com.redisgate.redis; import io.lettuce.core.RedisFuture; import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.pubsub.RedisPubSubAdapter; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands; import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RestController; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @RestController @Slf4j public class Redis10_PubSub { private final RedisConn redisConn; private final StatefulRedisPubSubConnection<String, String> redisPubConn; private final RedisPubSubCommands<String, String> redisPubCommands; private final RedisPubSubAsyncCommands<String, String> redisPubAsyncCommands; private final StatefulRedisPubSubConnection<String, String> redisSubConn; private final RedisPubSubCommands<String, String> redisSubCommands; private final RedisPubSubAsyncCommands<String, String> redisSubAsyncCommands; public Redis10_PubSub(RedisConn redisConn) { this.redisConn = redisConn; // Publish connection redisPubConn = redisConn.getRedisPubConn(); redisPubCommands = redisPubConn.sync(); redisPubAsyncCommands = redisPubConn.async(); // Subscribe connection redisSubConn = redisConn.getRedisSubConn(); redisSubCommands = redisSubConn.sync(); redisSubAsyncCommands = redisSubConn.async(); } // PubSub Connection에 Listener를 등록한다. // 여기서 subscribe 한다. // http://localhost:8080/pubsubAddListener @GetMapping("/pubsubAddListener") public String pubsubAddListener() { // PubSub 메시지 수신 redisSubConn.addListener(new RedisPubSubAdapter<String, String>() { // 이 메서드는 Publish 명령 실행 시 실행된다. // Channel으로 수신한다. @Override public void message(String channel, String message) { System.out.println("Subscribe(channel) "+channel+" "+message); } // 이 메서드는 Publish 명령 실행 시 실행된다. // Pattern으로 수신한다. @Override public void message(String pattern, String channel, String message) { System.out.println("Subscribe(pattern) "+pattern+" "+channel+" "+message); } // 이 메서드는 subscribe 명령 실행 시 실행된다. // count: 수신(subscribe) 등록된 channel 개수(아래 패턴으로 등록된 개수 포함) @Override public void subscribed(String channel, long count) { System.out.println("Subscribe(count) "+channel+" "+count); } // 이 메서드는 psubscribe 명령 실행 시 실행된다. // count: 패턴 수신(psubscribe) 등록된 channel 개수(위 채널로 등록된 개수 포함) @Override public void psubscribed(String pattern, long count) { System.out.println("Psubscribe(count) "+pattern+" "+count); } }); String msg = "OK - pubsubAddListener"; System.out.println(msg); return msg; } /* 1. Application 실행 중 서버가 다운되었다 다시 시작하면 Lettuce에서 자동으로 subscribe 명령을 다시 실행한다. PubSubSync, PubSubAsync 동일하다. 주의: Master/Replica 구성에서 마스터가 바뀌면(failover) 적용되지 않는다. Redis Server Log 11:24:50.590 - Accepted 192.168.56.1:11978 11:24:50.590 - 192.168.56.1:11978: HELLO 3 AUTH default redisgate SETNAME redisgate 11:24:50.596 - 192.168.56.1:11978: SUBSCRIBE ch01 ch02 2. Subscribe(등록)는 Master/Replica 구성에서 Failover 시 새 마스터에 자동으로 전파(등록)되지 않습니다. 새로 Subscribe(등록)해야 합니다. 3. 아래 코드와 같이 While()을 사용할 필요가 없다. - AI답변/인터넷에 이런 예제가 있다. Redis는 subscribe 명령을 한번만 실행하면 된다. 즉, subscribe 명령은 등록하는 것으로 한번만 등록하면 된다. Thread subscriberThread = new Thread(() -> { while (true) { redisPubSubCommands.subscribe(channel); } }); subscriberThread.start(); */ // Sync Subscribe // SUBSCRIBE channel [channel ...] // http://localhost:8080/subscribe/ch01 @GetMapping("/subscribe/{channel}") public String subscribe(@PathVariable("channel") String channel) { try { redisSubCommands.subscribe(channel); } catch (Exception e) { System.out.println(e.getMessage()); return "Error - subscribe "+channel; } return "OK - subscribe "+channel; } // Async Subscribe // SUBSCRIBE channel [channel ...] // http://localhost:8080/subscribeAsync/ch02 @GetMapping("/subscribeAsync/{channel}") public String subscribeAsync(@PathVariable("channel") String channel) { RedisFuture<Void> future = redisSubAsyncCommands.subscribe(channel); try { future.get(5, TimeUnit.SECONDS); } catch (Exception e) { System.out.println(e.getMessage()); return "Error - Async subscribe "+channel; } return "OK - Async subscribe "+channel; } // Pattern subscribe // PSUBSCRIBE pattern [pattern ...] // http://localhost:8080/subscribePattern/ch* @GetMapping("/subscribePattern/{pattern}") public String subscribePattern(@PathVariable("pattern") String pattern) { RedisFuture<Void> future = redisSubAsyncCommands.psubscribe(pattern); try { future.get(5, TimeUnit.SECONDS); } catch (Exception e) { System.out.println(e.getMessage()); return "Error - Async psubscribe "+pattern; } return "OK - Async psubscribe "+pattern; } /** * Publish channel message * @param channelMsg channel:message로 구성됩니다. * @return "Publish channel message" */ // PUBLISH channel message // http://localhost:8080/publish/ch01:Hello @GetMapping("/publish/{channelMsg}") public String publish(@PathVariable("channelMsg") String channelMsg) { String[] msg = channelMsg.split(":"); String result = "Publish "+msg[0]+" "+msg[1]; System.out.println(result); Long num = redisPubCommands.publish(msg[0], msg[1]); return result+" -> "+num; } // Channel list를 조회한다. // PUBSUB channels // http://localhost:8080/pubsubChannels @GetMapping("/pubsubChannels") public List<String> pubsubChannels() { List<String> pubsubChannels = redisPubCommands.pubsubChannels(); pubsubChannels.forEach(System.out::println); return pubsubChannels; } // 해당 채널에 등록된 클라이언트 개수를 보여준다. // PUBSUB numsub [channel-1 ... channel-n] // http://localhost:8080/pubsubNumsub @GetMapping("/pubsubNumsub") public Map<String, Long> pubsubNumsub() { Map<String, Long> pubsubNumsub = redisPubCommands.pubsubNumsub("ch01","ch02"); for (Map.Entry<String, Long> entry : pubsubNumsub.entrySet()) { String channel = entry.getKey(); Long clients = entry.getValue(); System.out.println("Channel: " + channel + ", Clients: " + clients); } return pubsubNumsub; } // 서버에 등록된 pattern의 개수를 보여준다. // PUBSUB numpat // http://localhost:8080/pubsubNumpat @GetMapping("/pubsubNumpat") public Long pubsubNumpat() { Long pubsubNumpat = redisPubCommands.pubsubNumpat(); System.out.println("pubsubNumpat: "+pubsubNumpat); return pubsubNumpat; } }

각 명령(메서드) 별 표시

SUBSCRIBE

// Sync Subscribe // SUBSCRIBE channel [channel ...] // http://localhost:8080/subscribe/ch01 @GetMapping("/subscribe/{channel}") public String subscribe(@PathVariable("channel") String channel) { try { redisSubCommands.subscribe(channel); } catch (Exception e) { System.out.println(e.getMessage()); return "Error - subscribe "+channel; } return "OK - subscribe "+channel; } // Async Subscribe // SUBSCRIBE channel [channel ...] // http://localhost:8080/subscribeAsync/ch02 @GetMapping("/subscribeAsync/{channel}") public String subscribeAsync(@PathVariable("channel") String channel) { RedisFuture<Void> future = redisSubAsyncCommands.subscribe(channel); try { future.get(5, TimeUnit.SECONDS); } catch (Exception e) { System.out.println(e.getMessage()); return "Error - Async subscribe "+channel; } return "OK - Async subscribe "+channel; }

PSUBSCRIBE

// Pattern subscribe // PSUBSCRIBE pattern [pattern ...] // http://localhost:8080/subscribePattern/ch* @GetMapping("/subscribePattern/{pattern}") public String subscribePattern(@PathVariable("pattern") String pattern) { RedisFuture<Void> future = redisSubAsyncCommands.psubscribe(pattern); try { future.get(5, TimeUnit.SECONDS); } catch (Exception e) { System.out.println(e.getMessage()); return "Error - Async psubscribe "+pattern; } return "OK - Async psubscribe "+pattern; }

PUBLISH

// PUBLISH channel message // http://localhost:8080/publish/ch01:Hello @GetMapping("/publish/{channelMsg}") public String publish(@PathVariable("channelMsg") String channelMsg) { String[] msg = channelMsg.split(":"); String result = "Publish "+msg[0]+" "+msg[1]; System.out.println(result); Long num = redisPubCommands.publish(msg[0], msg[1]); return result+" -> "+num; }

PUBSUB channels

// Channel list를 조회한다. // PUBSUB channels // http://localhost:8080/pubsubChannels @GetMapping("/pubsubChannels") public List<String> pubsubChannels() { List<String> pubsubChannels = redisPubCommands.pubsubChannels(); pubsubChannels.forEach(System.out::println); return pubsubChannels; }

PUBSUB numsub

// 해당 채널에 등록된 클라이언트 개수를 보여준다. // PUBSUB numsub [channel-1 ... channel-n] // http://localhost:8080/pubsubNumsub @GetMapping("/pubsubNumsub") public Map<String, Long> pubsubNumsub() { Map<String, Long> pubsubNumsub = redisPubCommands.pubsubNumsub("ch01","ch02"); for (Map.Entry<String, Long> entry : pubsubNumsub.entrySet()) { String channel = entry.getKey(); Long clients = entry.getValue(); System.out.println("Channel: " + channel + ", Clients: " + clients); } return pubsubNumsub; }

PUBSUB numpat

// 서버에 등록된 pattern의 개수를 보여준다. // PUBSUB numpat // http://localhost:8080/pubsubNumpat @GetMapping("/pubsubNumpat") public Long pubsubNumpat() { Long pubsubNumpat = redisPubCommands.pubsubNumpat(); System.out.println("pubsubNumpat: "+pubsubNumpat); return pubsubNumpat; }