如何使用 longadder 执行

Question

您好，我有一个应用程序计算时间并据此引发事件。我想使用 longadder 或任何合适的方法使我的应用程序具有线程安全性。

下面是我的班级；

    @Autowired
    EventListenerConfiguration eventListenerConfiguration;

    private volatile long lastReceivedMessage = System.currentTimeMillis();

 
    public void consume(String message) Integer partition,
                        (Headers.OFFSET) Long offset, Acknowledgment ack) {


        lastReceivedMessage = System.currentTimeMillis();

        try {
            seervice.processMessage(message, ack, null);

        } catch (ParseException e) {
            logger.error(e.getMessage());
        }
    }

    @Scheduled(fixedDelayString = "${listenScheduled}", initialDelay = 100000)
    private void distanceBetweenLastReceivedMessageAndCurrentTime() {

        long currentTime = System.currentTimeMillis() - lastReceivedMessage;

        if (currentTime >= EventListenerConfiguration .getTotalMilliSecondTimeForError()) {

            EventUtil.publishEvent(THROW_ERROR_EVENT, EventSeverityStatus.ERROR, EventTypeStatus.CUSTOM, null);


        } else (currentTime >= EventListenerConfiguration.getTotalMilliSecondTimeForWarn()) {

            EventUtil.publishEvent(THROW_WARN_EVENT, EventSeverityStatus.WARN, EventTypeStatus.CUSTOM, null);

        }
}

所以基本上如何在不改变 longAdder 的情况下转换我的代码并执行 currentTime-lastReceiveMessage

谢谢

Answer 1

晚上好。看来您的问题主要是 lastReceivedMessage 是两个线程使用的受保护资源。无论运行什么，consume 方法都会生成它，然后 Spring-Generated @scheduled 线程会消耗它。

添加 volatile 关键字不会阻止代码读取该字段。它只会阻止代码缓存变量。 （继续阅读 volatile https://www.geeksforgeeks.org/volatile-keyword-in-java/）如果您想将 @Scheduled 块视为关键区域，并在完成之前阻止 lastRecievedMessage 的更新，我建议您执行以下操作：

    private volatile long lastReceivedMessage = System.currentTimeMillis();
    private Semaphore resourceLock = new Semaphore(1);

    public void consume(String message) Integer partition,
    (Headers.OFFSET) Long offset, Acknowledgment ack) {

        resourceLock.acquireUninterruptibly();
        try {
            lastReceivedMessage = System.currentTimeMillis();
        } finally {
            resourceLock.release();
        }

        try {
            seervice.processMessage(message, ack, null);

        } catch (ParseException e) {
            logger.error(e.getMessage());
        }
    }

    @Scheduled(fixedDelayString = "${listenScheduled}", initialDelay = 100000)
    private void distanceBetweenLastReceivedMessageAndCurrentTime() {
        long currentTime = 0;

        resourceLock.acquireUninterruptibly();
        try {
                currentTime = System.currentTimeMillis() - lastReceivedMessage;
                
            if (currentTime >= EventListenerConfiguration .getTotalMilliSecondTimeForError()) {
                EventUtil.publishEvent(THROW_ERROR_EVENT, EventSeverityStatus.ERROR, EventTypeStatus.CUSTOM, null);

            } else (currentTime >= EventListenerConfiguration.getTotalMilliSecondTimeForWarn()) {
                EventUtil.publishEvent(THROW_WARN_EVENT, EventSeverityStatus.WARN, EventTypeStatus.CUSTOM, null);
            }
        } finally {
            resourceLock.release();
        }
    }