Spring Boot中通过KafkaListener更新共享Prices对象的实现是否安全？

我正在实现一个Spring Boot应用，该应用从Kafka读取数据并提供给所有请求客户端。初始的DataProvider类如下：

@Component
public class DataProvider {
    
    private Prices prices;
    
    public DataProvider() {
        this.prices = Prices.of();
    }
    
    public Prices getPrices() {
        return prices;
    }
}

客户端可通过GET /api/prices获取最新价格信息。价格的实时更新从Kafka消费而来，因更新频率为每5秒一次（不算频繁），该Kafka主题仅设一个分区。我尝试了使用Kafka Listener的基础实现版本：

@Component
public class DataProvider {

    private Prices prices;

    public DataProvider() {
        this.prices = Prices.of();
    }

    public Prices getPrices() {
        return prices;
    }

    @KafkaListener(topics = "test-topic")
    public void consume(String message) {
        Prices prices = Prices.of(message);
        this.prices = prices;
    }
}

请问这种实现方式是否安全？

这种实现不是线程安全的，核心问题在于多线程环境下的可见性：

• 处理HTTP请求的线程（来自Spring Boot内嵌容器的线程池）和Kafka消费线程是完全独立的线程。当前代码中prices是普通成员变量，当消费线程更新prices的引用时，HTTP线程可能无法立即看到最新的值，甚至永远看不到——这是Java内存模型中线程私有工作内存的缓存机制导致的可见性问题。
• 虽然更新频率低（5秒一次），但只要存在多线程并发读写，就存在用户请求获取旧价格数据的风险，违背了“提供最新价格”的需求。

修复方案 #

因为你的Kafka主题只有一个分区，默认情况下@KafkaListener只会启动一个消费线程，所以不存在多个线程同时更新prices的并发冲突问题，只需要解决可见性即可，有两种简单可靠的方式：

1. 用volatile修饰prices变量

volatile关键字能保证变量的可见性，禁止指令重排序，确保消费线程更新的prices引用能立即被所有HTTP线程看到：

@Component
public class DataProvider {

    private volatile Prices prices;

    public DataProvider() {
        this.prices = Prices.of();
    }

    public Prices getPrices() {
        return prices;
    }

    @KafkaListener(topics = "test-topic")
    public void consume(String message) {
        Prices prices = Prices.of(message);
        this.prices = prices;
    }
}

2. 用AtomicReference包装Prices对象

AtomicReference是Java并发包提供的原子类，它内部通过volatile和CAS操作保证了引用更新的原子性和可见性，适合这种单线程更新、多线程读取的场景：

@Component
public class DataProvider {

    private final AtomicReference<Prices> pricesRef;

    public DataProvider() {
        this.pricesRef = new AtomicReference<>(Prices.of());
    }

    public Prices getPrices() {
        return pricesRef.get();
    }

    @KafkaListener(topics = "test-topic")
    public void consume(String message) {
        Prices newPrices = Prices.of(message);
        pricesRef.set(newPrices);
    }
}

额外说明 #

如果未来你的Kafka主题扩展为多个分区，@KafkaListener会启动多个消费线程（默认一个分区对应一个线程），此时如果需要保证更新的顺序性（比如避免旧价格覆盖新价格），可能需要额外的同步机制，但就当前单分区场景而言，上面两种方案完全足够。

内容的提问来源于stack exchange，提问作者Michal Kowalski