संदेश को फ़्लश करने के लिए मैं AmazonSQSBufferedAsyncClient को कैसे मजबूर कर सकता हूं?

Question

मैं जावा के लिए एडब्ल्यूएस एसडीके का उपयोग कर रहा हूं और मैं बैच अनुरोधों के लिए बफरिंग एसिंक एसक्यू क्लाइंट का उपयोग कर रहा हूं ताकि मैं लागत कम कर सकूं।

जब मेरा एप्लिकेशन बंद हो जाता है, तो मैं यह सुनिश्चित करना चाहता हूं कि बफर में कोई संदेश प्रतीक्षा नहीं कर रहा है, लेकिन क्लाइंट पर कोई .flush() विधि नहीं है।

क्या AmazonSQSBufferedAsyncClient.shutdown() कॉल किए जाने पर मेरे संदेश फ़्लश करते हैं? मैंने source code देखा और यह अस्पष्ट है। प्रणाली को बुलाती है प्रत्येक QueueBuffer यह है कि पर shutdown(), लेकिन QueueBuffer.shutdown() अंदर यह

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

आगे कहते हैं, .shutdown() के लिए दस्तावेज़ का कहना है:

बन्द हो जाता है इस ग्राहक वस्तु नीचे, किसी भी संसाधनों को रिहा हो सकता है कि खुला यह एक वैकल्पिक विधि है, और कॉलर्स को पर कॉल करने की उम्मीद नहीं है, लेकिन अगर वे स्पष्ट रूप से किसी भी खुले संसाधनों को रिलीज़ करना चाहते हैं। एक बार ग्राहक बंद हो जाने के बाद, इसका उपयोग पर और अधिक अनुरोध करने के लिए नहीं किया जाना चाहिए।

इस एप्लिकेशन के लिए, मुझे यह सुनिश्चित करने की ज़रूरत है कि buffered होने पर कोई संदेश खो जाए। क्या मुझे बफरिंग/async एक के बजाय सामान्य AmazonSQSClient का उपयोग करके मैन्युअल रूप से इसे संभालने की आवश्यकता है?

Answer 1

एसडीके के 1.11.37 संस्करण के साथ, QueueBufferConfig में इस उद्देश्य के लिए कॉन्फ़िगरेशन पैरामीटर है।

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

Answer 2

फ़्लश को स्पष्ट रूप से कॉल करने का एक तरीका है लेकिन यह सुलभ नहीं है और वास्तव में मुझे उस विधि को अमेज़ॅन कोड में कोई कॉल नहीं मिला। ऐसा लगता है कि कुछ गुम है।

जब आप async ग्राहक यह निम्नलिखित कोड निष्पादित पर बंद फोन:

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

और QueueBuffer # बंद() इस तरह दिखता है:

/** 
* 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(); 
} 

तो ऐसा लगता है जैसे वे जानबूझकर नहीं हैं भेजने के लिए बुला रहा है Buffer.shutdown() वह तरीका है जो बफर में हर संदेश को फ्लश करेगा जो अभी भी नहीं भेजा गया है।

क्या आपको एसक्यूएस क्लाइंट को बंद करने पर यह मामला मिला और यह संदेश खो गया? ऐसा लगता है कि वे इसके बारे में जानते हैं और वह मामला नहीं होना चाहिए, लेकिन यदि आप यह सुनिश्चित करना चाहते हैं कि आप उस विधि को प्रतिबिंब के साथ बुला सकते हैं जो वास्तव में बुरा है लेकिन यह आपकी आवश्यकताओं को पूरा करेगा।

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 
    } 

ऐसा कुछ काम करना चाहिए, मुझे लगता है। मुझे बताएं!