class RedissonBroadcastManager extends BroadcastManager {
    private static final Logger logger = LoggerFactory.getLogger(RedissonBroadcastManager.class);
    private final RedissonCacheConfig<?, ?> config;
    private final String channel;
    private final RedissonClient client;
    private volatile int subscribeId;

    private final ReentrantLock reentrantLock = new ReentrantLock();

    public RedissonBroadcastManager(final CacheManager cacheManager, final RedissonCacheConfig<?, ?> config) {
        super(cacheManager);
        checkConfig(config);
        this.config = config;
        this.channel = config.getBroadcastChannel();
        this.client = config.getRedissonClient();
    }

    @Override
    public void startSubscribe() {
        reentrantLock.lock();
        try {
            if (this.subscribeId == 0 && Objects.nonNull(this.channel) && !this.channel.isEmpty()) {
                this.subscribeId = this.client.getTopic(this.channel)
                        .addListener(byte[].class, (channel, msg) -> processNotification(msg, this.config.getValueDecoder()));
            }
        }finally {
            reentrantLock.unlock();
        }
    }


    @Override
    public void close() {<FILL_FUNCTION_BODY>}

    @Override
    public CacheResult publish(final CacheMessage cacheMessage) {
        try {
            if (Objects.nonNull(this.channel) && Objects.nonNull(cacheMessage)) {
                final byte[] msg = this.config.getValueEncoder().apply(cacheMessage);
                this.client.getTopic(this.channel).publish(msg);
                return CacheResult.SUCCESS_WITHOUT_MSG;
            }
            return CacheResult.FAIL_WITHOUT_MSG;
        } catch (Throwable e) {
            SquashedLogger.getLogger(logger).error("jetcache publish error", e);
            return new CacheResult(e);
        }
    }
}

        reentrantLock.lock();
        try {
            final int id;
            if ((id = this.subscribeId) > 0 && Objects.nonNull(this.channel)) {
                this.subscribeId = 0;
                try {
                    this.client.getTopic(this.channel).removeListener(id);
                } catch (Throwable e) {
                    logger.warn("unsubscribe {} fail", this.channel, e);
                }
            }
        }finally {
            reentrantLock.unlock();
        }
    
<methods>public void <init>(com.alicp.jetcache.CacheManager) ,public void close() throws java.lang.Exception,public com.alicp.jetcache.CacheManager getCacheManager() ,public java.lang.String getSourceId() ,public abstract com.alicp.jetcache.CacheResult publish(com.alicp.jetcache.support.CacheMessage) ,public abstract void startSubscribe() <variables>private final non-sealed com.alicp.jetcache.CacheManager cacheManager,private static Logger logger,private final java.lang.String sourceId