class SpringDataBroadcastManager extends BroadcastManager {

    private static final Logger logger = LoggerFactory.getLogger(SpringDataBroadcastManager.class);

    private final RedisSpringDataCacheConfig config;
    private final MessageListener listener = this::onMessage;
    private final byte[] channel;
    private volatile RedisMessageListenerContainer listenerContainer;

    private final ReentrantLock reentrantLock = new ReentrantLock();

    public SpringDataBroadcastManager(CacheManager cacheManager, RedisSpringDataCacheConfig config) {
        super(cacheManager);
        this.config = config;
        checkConfig(config);
        if (config.getConnectionFactory() == null) {
            throw new CacheConfigException("connectionFactory is required");
        }
        this.channel = config.getBroadcastChannel().getBytes(StandardCharsets.UTF_8);
    }

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

    @Override
    public void startSubscribe() {
        reentrantLock.lock();
        try {
            if (this.listenerContainer != null) {
                throw new IllegalStateException("subscribe thread is started");
            }
            Topic topic = new ChannelTopic(config.getBroadcastChannel());
            if (config.getListenerContainer() == null) {
                RedisMessageListenerContainer c = new RedisMessageListenerContainer();
                c.setConnectionFactory(config.getConnectionFactory());
                c.afterPropertiesSet();
                c.start();
                this.listenerContainer = c;
                logger.info("create RedisMessageListenerContainer instance");
            } else {
                this.listenerContainer = config.getListenerContainer();
            }
            this.listenerContainer.addMessageListener(listener, topic);
            logger.info("subscribe jetcache invalidate notification. channel={}", config.getBroadcastChannel());
        }finally {
            reentrantLock.unlock();
        }
    }

    private void onMessage(Message message, byte[] pattern) {
        processNotification(message.getBody(), config.getValueDecoder());
    }

    @Override
    public void close() throws Exception {
        reentrantLock.lock();
        try {
            if (this.listenerContainer != null) {
                this.listenerContainer.removeMessageListener(listener);
                if (this.config.getListenerContainer() == null) {
                    this.listenerContainer.destroy();
                }
            }
            this.listenerContainer = null;
        }finally {
            reentrantLock.unlock();
        }
    }
}

        RedisConnection con = null;
        try {
            con = config.getConnectionFactory().getConnection();
            byte[] body = (byte[]) config.getValueEncoder().apply(cacheMessage);
            con.publish(channel, body);
            return CacheResult.SUCCESS_WITHOUT_MSG;
        } catch (Exception ex) {
            SquashedLogger.getLogger(logger).error("jetcache publish error", ex);
            return new CacheResult(ex);
        } finally {
            if (con != null) {
                try {
                    con.close();
                } catch (Exception e) {
                    SquashedLogger.getLogger(logger).error("RedisConnection close fail", e);
                }
            }
        }
    
<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