Làm thế nào tôi có thể buộc AmazonSQSBufferedAsyncClient xóa thư?

Question

Tôi đang sử dụng SDK AWS cho Java và tôi đang sử dụng trình khách vùng đệm async đệm để yêu cầu hàng loạt để tôi giảm chi phí.

Khi ứng dụng của tôi tắt, tôi muốn đảm bảo rằng không có thư nào đang chờ trong bộ đệm, nhưng không có phương thức .flush() tôi có thể thấy trên máy khách.

Có AmazonSQSBufferedAsyncClient.shutdown() xóa thư khi được gọi? Tôi nhìn vào số source code và không rõ ràng. Phương pháp này gọi shutdown() trên mỗi QueueBuffer rằng nó có, nhưng bên trong QueueBuffer.shutdown() nó nói

public void shutdown() { 
    //send buffer does not require shutdown, only 
    //shut down receive buffer 
    receiveBuffer.shutdown(); 
} 

Hơn nữa, tài liệu cho .shutdown() nói:

tắt đối tượng khách hàng này, phát hành bất kỳ tài nguyên mà có thể tổ chức mở. Đây là phương pháp tùy chọn và người gọi không được mong đợi để gọi số, nhưng có thể nếu họ muốn phát hành rõ ràng mọi tài nguyên mở . Khi một khách hàng đã bị tắt, nó không nên được sử dụng để thực hiện thêm bất kỳ yêu cầu nào.

Đối với ứng dụng này, tôi cần đảm bảo không có thư bị mất trong khi được lưu vào bộ đệm. Tôi có cần phải xử lý thủ công điều này bằng cách sử dụng thông thường AmazonSQSClient thay vì sử dụng chế độ đệm/không đồng bộ không?

Answer 1

Với phiên bản 1.11.37 của SDK, có thông số cấu hình chỉ dành cho mục đích này trong QueueBufferConfig.

AmazonSQSBufferedAsyncClient bufClient = 
    new AmazonSQSBufferedAsyncClient(
     realAsyncClient, 
     new QueueBufferConfig() 
      .withFlushOnShutdown(true) 
    ); 

Answer 2

Có một phương pháp để gọi rõ ràng lệnh xả nhưng không truy cập được và thực sự tôi không thể tìm thấy bất kỳ lệnh gọi nào đến phương thức đó trong mã amazon. Nó có vẻ như một cái gì đó là mất tích.

Khi bạn gọi tắt trên máy khách async nó thực thi đoạn mã sau:

public void shutdown() { 
    for(QueueBuffer buffer : buffers.values()) { 
     buffer.shutdown(); 
    } 
    realSQS.shutdown(); 
} 

Và QueueBuffer # shutdown() trông như thế này:

/** 
* Shuts down the queue buffer. Once this method has been called, the 
* queue buffer is not operational and all subsequent calls to it may fail 
* */ 
public void shutdown() { 
    //send buffer does not require shutdown, only 
    //shut down receive buffer 
    receiveBuffer.shutdown(); 
} 

Vì vậy, nó có vẻ như họ đang cố tình không gọi đến sendBuffer.shutdown() là phương thức sẽ xóa mọi thư trong bộ đệm mà vẫn chưa được gửi.

Bạn có tìm thấy trường hợp khi tắt máy khách SQS và mất thư không? Dường như họ nhận thức được điều đó và trường hợp đó không nên xảy ra, nhưng nếu bạn muốn chắc chắn bạn có thể gọi phương thức đó với sự phản chiếu mà nó thực sự khó chịu nhưng nó sẽ thỏa mãn nhu cầu của bạn.

AmazonSQSBufferedAsyncClient asyncSqsClient = <your initialization code of the client>; 
    Field buffersField = ReflectionUtils.findField(AmazonSQSBufferedAsyncClient.class, "buffers"); 
    ReflectionUtils.makeAccessible(buffersField); 
    LinkedHashMap<String, Object> buffers = (LinkedHashMap<String, Object>) ReflectionUtils.getField(buffersField, asyncSqsClient); 
    for (Object buffer : buffers.values()) { 
     Class<?> clazz = Class.forName("com.amazonaws.services.sqs.buffered.QueueBuffer"); 
     SendQueueBuffer sendQueueBuffer = (SendQueueBuffer) ReflectionUtils.getField(ReflectionUtils.findField(clazz, "sendBuffer"), buffer); 
     sendQueueBuffer.flush();//finally 
    } 

Điều gì đó tương tự sẽ hiệu quả, tôi đoán vậy. Cho tôi biết!