redis发布、订阅

@Setter
public final class RedisUserLoginRegisterSubscribe extends JedisPubSub {
private static final Logger          log             = LoggerFactory.getLogger(REDIS_SUBPUB_LOGGER_NAME);
/**
* 订阅处理
*/
private final static ExecutorService executorService = Executors.newFixedThreadPool(32);

/**
* Redis 连接池
*/
private JedisPool                           jedisPool;
/**
* 用户注册订阅处理器
*/
private UserLoginRegisterSubscribeProcessor userRegisterSubscribeProcessor;

private boolean isSubscribe = true;

public void init() {
executorService.execute(() -> {
log.info("初始化用户注册订阅...");
while (isSubscribe) {
try (Jedis jedis = jedisPool.getResource()) {
jedis.subscribe(this, USER_REGISTER_PUBSUB_CHANNEL);
} catch (Exception ex) {
log.error(ex.getMessage(),ex);
}
try {
Thread.sleep(500);
} catch (InterruptedException e) {
}
}
});
}

public void destroy() {
log.info("用户注册订阅开始销毁处理...");
if (!executorService.isShutdown()) {
executorService.shutdown();
}
log.info("用户注册订阅销毁处理成功...");
}

@Override
public void onMessage(String channel, String message) {
executorService.execute(() -> {
log.info("收到消息 channel:{}, message:{}", channel, message);
UserLoginRegisterMessage userRegisterMessage = JSONObject.parseObject(message, UserLoginRegisterMessage.class);
userRegisterMessage.setChannel(channel);
userRegisterSubscribeProcessor.process(userRegisterMessage);
});
}

@Override
public void onPUnsubscribe(String pattern, int subscribedChannels) {
isSubscribe = false;
}

@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
isSubscribe = false;
}
}

public interface UserLoginRegisterSubscribeProcessor {

/**
* 订阅接受到用户注册消息的处理
*
* @param message {@link UserLoginRegisterMessage}
*/
void process(UserLoginRegisterMessage message);
}

rs = jedis.publish(USER_REGISTER_PUBSUB_CHANNEL, JSON.toJSONString(userRegisterMessage));

* <bean id="userRegisterSubscribeProcessor" class="...实现类"/>
*
* <bean id="redisUserRegisterSubscribe" init-method="init" destroy-method="destroy"
*         class="com.***.usercore.sdk.pubsub.RedisUserLoginRegisterSubscribe">
*     <property name="jedisPool" ref="jedisPool"/>
*     <property name="userRegisterSubscribeProcessor" ref="userRegisterSubscribeProcessor"/>
* </bean>