class LettuceBroadcastManager extends BroadcastManager {
    private static final Logger logger = LoggerFactory.getLogger(LettuceBroadcastManager.class);

    private final RedisLettuceCacheConfig<Object, Object> config;
    private final byte[] channel;

    private volatile boolean subscribeThreadStart;
    private volatile RedisPubSubAdapter<byte[], byte[]> pubSubAdapter;

    private final LettuceConnectionManager lettuceConnectionManager;
    private final BaseRedisAsyncCommands<byte[], byte[]> stringAsyncCommands;

    private final ReentrantLock reentrantLock = new ReentrantLock();


    public LettuceBroadcastManager(CacheManager cacheManager, RedisLettuceCacheConfig<Object, Object> config) {
        super(cacheManager);
        checkConfig(config);
        if (config.getPubSubConnection() == null) {
            throw new CacheConfigException("PubSubConnection not set");
        }

        this.config = config;
        this.channel = config.getBroadcastChannel().getBytes(StandardCharsets.UTF_8);
        this.lettuceConnectionManager = config.getConnectionManager();
        this.lettuceConnectionManager.init(config.getRedisClient(), config.getConnection());
        this.stringAsyncCommands = (BaseRedisAsyncCommands<byte[], byte[]>) lettuceConnectionManager
                .asyncCommands(config.getRedisClient());
    }

    @Override
    public CacheResult publish(CacheMessage cacheMessage) {<FILL_FUNCTION_BODY>}

    @Override
    public void startSubscribe() {
        reentrantLock.lock();
        try {
            if (subscribeThreadStart) {
                throw new IllegalStateException("startSubscribe has invoked");
            }
            this.pubSubAdapter = new RedisPubSubAdapter<byte[], byte[]>() {
                @Override
                public void message(byte[] channel, byte[] message) {
                    processNotification(message, config.getValueDecoder());
                }
            };
            config.getPubSubConnection().addListener(this.pubSubAdapter);
            RedisPubSubAsyncCommands<byte[], byte[]> asyncCommands = config.getPubSubConnection().async();
            asyncCommands.subscribe(channel);
            this.subscribeThreadStart = true;
        }finally {
            reentrantLock.unlock();
        }
    }

    @Override
    public void close() {
        config.getPubSubConnection().removeListener(this.pubSubAdapter);
        config.getPubSubConnection().close();
    }
}

        try {
            byte[] value = config.getValueEncoder().apply(cacheMessage);
            RedisFuture<Long> future = stringAsyncCommands.publish(channel, value);
            return new CacheResult(future.handle((rt, ex) -> {
                if (ex != null) {
                    JetCacheExecutor.defaultExecutor().execute(() ->
                            SquashedLogger.getLogger(logger).error("jetcache publish error", ex));
                    return new ResultData(ex);
                } else {
                    return new ResultData(CacheResultCode.SUCCESS, null, null);
                }
            }));
        } catch (Exception ex) {
            SquashedLogger.getLogger(logger).error("jetcache publish error", ex);
            return new CacheResult(ex);
        }
    
<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