java 異步查詢轉同步多種實現方式：循環等待，CountDownLatch，EventListener

時間 2019-12-07

標籤 java 異步查詢同步多種實現方式循環等待 countdownlatch eventlistener 欄目 Java 简体版

原文原文鏈接

異步轉同步

業務需求

有些接口查詢反饋結果是異步返回的，沒法馬上獲取查詢結果。java

正常處理邏輯

觸發異步操做，而後傳遞一個惟一標識。git

等到異步結果返回，根據傳入的惟一標識，匹配這次結果。github

如何轉換爲同步

正常的應用場景不少，可是有時候不想作數據存儲，只是想簡單獲取調用結果。spring

即想達到同步操做的結果，怎麼辦呢？apache

思路

發起異步操做
在異步結果返回以前，一直等待(能夠設置超時)
結果返回以後，異步操做結果統一返回

循環等待

LoopQuery.java

使用 query()，將異步的操做 remoteCallback() 執行完成後，同步返回。異步

public class LoopQuery implements Async {

    private String result;

    private static final Logger LOGGER = LogManager.getLogger(LoopQuery.class.getName());

    @Override
    public String query(String key) {
        startQuery(key);
        new Thread(new Runnable() {
            @Override
            public void run() {
                remoteCallback(key);
            }
        }).start();

        final String queryResult = endQuery();
        LOGGER.info("查詢結果: {}", queryResult);
        return queryResult;
    }

    /**
     * 開始查詢
     * @param key 查詢條件
     */
    private void startQuery(final String key) {
        LOGGER.info("執行查詢: {}", key);
    }

    /**
     * 遠程的回調是等待是隨機的
     *
     * @param key 查詢條件
     */
    private void remoteCallback(final String key) {
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.result = key + "-result";
        LOGGER.info("remoteCallback set result: {}", result);
    }

    /**
     * 結束查詢
     * @return 返回結果
     */
    private String endQuery() {
        while (true) {
            if (null == result) {
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                return result;
            }
        }
    }
}

main()

public static void main(String[] args) {
    new LoopQuery().query("12345");
}

測試結果

18:14:16.491 [main] INFO  com.github.houbb.thread.learn.aysnc.loop.LoopQuery - 執行查詢: 12345
18:14:21.498 [Thread-1] INFO  com.github.houbb.thread.learn.aysnc.loop.LoopQuery - remoteCallback set result: 12345-result
18:14:21.548 [main] INFO  com.github.houbb.thread.learn.aysnc.loop.LoopQuery - 查詢結果: 12345-result

CountDownLatch

AsyncQuery.java

使用 CountDownLatch 類達到同步的效果。async

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class AsyncQuery {

    private static final Logger LOGGER = LogManager.getLogger(AsyncQuery.class.getName());

    /**
     * 結果
     */
    private String result;

    /**
     * 異步轉同步查詢
     * @param key
     */
    public void asyncQuery(final String key) {
        final CountDownLatch latch = new CountDownLatch(1);
        this.startQuery(key);

        new Thread(new Runnable() {
            @Override
            public void run() {
                LOGGER.info("遠程回調線程開始");
                remoteCallback(key, latch);
                LOGGER.info("遠程回調線程結束");
            }
        }).start();

        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        this.endQuery();
    }

    private void startQuery(final String key) {
        LOGGER.info("執行查詢: {}", key);
    }

    /**
     * 遠程的回調是等待是隨機的
     * @param key
     */
    private void remoteCallback(final String key, CountDownLatch latch) {
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.result = key + "-result";
        latch.countDown();
    }

    private void endQuery() {
        LOGGER.info("查詢結果: {}", result);
    }

}

main()

public static void main(String[] args) {
    AsyncQuery asyncQuery = new AsyncQuery();
    final String key = "123456";
    asyncQuery.asyncQuery(key);
}

日誌

18:19:12.714 [main] INFO  com.github.houbb.thread.learn.aysnc.countdownlatch.AsyncQuery - 執行查詢: 123456
18:19:12.716 [Thread-1] INFO  com.github.houbb.thread.learn.aysnc.countdownlatch.AsyncQuery - 遠程回調線程開始
18:19:17.720 [main] INFO  com.github.houbb.thread.learn.aysnc.countdownlatch.AsyncQuery - 查詢結果: 123456-result
18:19:17.720 [Thread-1] INFO  com.github.houbb.thread.learn.aysnc.countdownlatch.AsyncQuery - 遠程回調線程結束

Spring EventListener

使用觀察者模式也能夠。（對方案一的優化）ide

此處結合 spring 進行使用。oop

BookingCreatedEvent.java

定義一個傳輸屬性的對象。測試

public class BookingCreatedEvent extends ApplicationEvent {

    private static final long serialVersionUID = -1387078212317348344L;

    private String info;

    public BookingCreatedEvent(Object source) {
        super(source);
    }

    public BookingCreatedEvent(Object source, String info) {
        super(source);
        this.info = info;
    }

    public String getInfo() {
        return info;
    }
}

BookingService.java

說明：當 this.context.publishEvent(bookingCreatedEvent); 觸發時，
會被 @EventListener 指定的方法監聽到。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class BookingService {

    @Autowired
    private ApplicationContext context;

    private volatile BookingCreatedEvent bookingCreatedEvent;

    /**
     * 異步轉同步查詢
     * @param info
     * @return
     */
    public String asyncQuery(final String info) {
        query(info);

        new Thread(new Runnable() {
            @Override
            public void run() {
                remoteCallback(info);
            }
        }).start();

        while(bookingCreatedEvent == null) {
            //.. 空循環
            // 短暫等待。
            try {
                TimeUnit.MILLISECONDS.sleep(1);
            } catch (InterruptedException e) {
                //...
            }
            //2. 使用兩個單獨的 event...

        }

        final String result = bookingCreatedEvent.getInfo();
        bookingCreatedEvent = null;
        return result;
    }

    @EventListener
    public void onApplicationEvent(BookingCreatedEvent bookingCreatedEvent) {
        System.out.println("監聽到遠程的信息: " + bookingCreatedEvent.getInfo());
        this.bookingCreatedEvent = bookingCreatedEvent;
        System.out.println("監聽到遠程消息後: " + this.bookingCreatedEvent.getInfo());
    }

    /**
     * 執行查詢
     * @param info
     */
    public void query(final String info) {
        System.out.println("開始查詢: " + info);
    }

    /**
     * 遠程回調
     * @param info
     */
    public void remoteCallback(final String info) {
        System.out.println("遠程回調開始: " + info);

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 重發結果事件
        String result = info + "-result";
        BookingCreatedEvent bookingCreatedEvent = new BookingCreatedEvent(this, result);
        //觸發event
        this.context.publishEvent(bookingCreatedEvent);
    }
}

測試方法

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = SpringConfig.class)
public class BookServiceTest {

    @Autowired
    private BookingService bookingService;

    @Test
    public void asyncQueryTest() {
        bookingService.asyncQuery("1234");
    }

}

日誌

2018-08-10 18:27:05.958  INFO  [main] com.github.houbb.spring.lean.core.ioc.event.BookingService:84 - 開始查詢:1234
2018-08-10 18:27:05.959  INFO  [Thread-2] com.github.houbb.spring.lean.core.ioc.event.BookingService:93 - 遠程回調開始:1234
接收到信息: 1234-result
2018-08-10 18:27:07.964  INFO  [Thread-2] com.github.houbb.spring.lean.core.ioc.event.BookingService:73 - 監聽到遠程的信息: 1234-result
2018-08-10 18:27:07.964  INFO  [Thread-2] com.github.houbb.spring.lean.core.ioc.event.BookingService:75 - 監聽到遠程消息後: 1234-result
2018-08-10 18:27:07.964  INFO  [Thread-2] com.github.houbb.spring.lean.core.ioc.event.BookingService:106 - 已經觸發event
2018-08-10 18:27:07.964  INFO  [main] com.github.houbb.spring.lean.core.ioc.event.BookingService:67 - 查詢結果: 1234-result
2018-08-10 18:27:07.968  INFO  [Thread-1] org.springframework.context.support.GenericApplicationContext:993 - Closing org.springframework.context.support.GenericApplicationContext@5cee5251: startup date [Fri Aug 10 18:27:05 CST 2018]; root of context hierarchy

超時和空循環

空循環

空循環會致使 cpu 飆升

while(true) {
}

解決方式

while(true) {
    // 小睡便可
    TimeUnit.sleep(1);
}

超時編寫

不可能一直等待反饋，能夠設置超時時間。

/**
 * 循環等待直到獲取結果
 * @param key key
 * @param timeoutInSeconds 超時時間
 * @param <T> 泛型
 * @return 結果。若是超時則拋出異常
 */
public <T> T loopWaitForValue(final String key, long timeoutInSeconds) {
    long startTime = System.nanoTime();
    long deadline = startTime + TimeUnit.SECONDS.toNanos(timeoutInSeconds);
    //1. 若是沒有新回調，或者 key 對應元素不存在。則一直循環
    while(ObjectUtil.isNull(map.get(key))) {
        try {
            TimeUnit.MILLISECONDS.sleep(5);
        } catch (InterruptedException e) {
            LOGGER.warn("Loop meet InterruptedException, just ignore it.", e);
        }
        // 超時判斷
        long currentTime = System.nanoTime();
        if(currentTime >= deadline) {
            throw new BussinessException(ErrorCode.READ_TIME_OUT);
        }
    }
    final T target = (T) map.get(key);
    LOGGER.debug("loopWaitForValue get value:{} for key:{}", JSON.toJSON(target), key);
    //2. 獲取到元素以後，須要移除掉對應的值
    map.remove(key);
    return target;
}