<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

spring:
  redis:
    host: localhost
    port: 6379

@Component
public class RedisMessagePublisher {
    private final RedisTemplate<String, Object> redisTemplate;
 
    public void publish(String channel, TableChangeEvent event) {
        redisTemplate.convertAndSend(channel, event);
    }
}

@Component
public class RedisMessageSubscriber implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        TableChangeEvent event = deserialize(message.getBody());
        handleTableChange(event);
    }
}

@Data
@NoArgsConstructor
@AllArgsConstructor
public class TableChangeEvent {
    private String tableName;
    private ChangeType operation; // INSERT/UPDATE/DELETE
    private String entityId;
    private Instant changeTime;
}

@Configuration
public class RedisConfig {
    @Bean
    public RedisMessageListenerContainer container(
            RedisConnectionFactory factory,
            MessageListenerAdapter adapter) {
        
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener(adapter, new ChannelTopic("table_changes"));
        return container;
    }
}

@Service
public class ProductService {
    private final RedisMessagePublisher publisher;
 
    public Product saveProduct(Product product) {
        Product saved = repository.save(product);
        publisher.publish("table_changes", 
            new TableChangeEvent("products", ChangeType.INSERT, saved.getId()));
        return saved;
    }
}

@Bean
public StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamContainer(
        RedisConnectionFactory factory) {
    
    var options = StreamMessageListenerContainerOptions
        .builder()
        .pollTimeout(Duration.ofSeconds(1))
        .build();
    
    var container = StreamMessageListenerContainer.create(factory, options);
    
    container.receiveAutoAck(Consumer.from("app-group", "instance-1"),
        StreamOffset.create("table_changes_stream", ReadOffset.lastConsumed()),
        message -> processChange(message.getValue()));
    
    container.start();
    return container;
}

@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
    RedisTemplate<String, Object> template = new RedisTemplate<>();
    template.setConnectionFactory(factory);
    template.setValueSerializer(new Jackson2JsonRedisSerializer<>(Object.class));
    return template;
}

public void handleTableChange(TableChangeEvent event) {
    String lockKey = "lock:" + event.getTableName() + ":" + event.getEntityId();
    if (redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS)) {
        try {
            // 處理業(yè)務(wù)邏輯
        } finally {
            redisTemplate.delete(lockKey);
        }
    }
}