PushWithMultiThreadAndMultiEndpoint.java

public class PushWithMultiThreadAndMultiEndpoint {
  
    @Value("${cloud.aws.sns.endpoint}")
    private String SNS_ENDPOINT;

    @Value("${cloud.aws.credentials.accessKey}")
    private String ACCESS_KEY;

    @Value("${cloud.aws.credentials.secretKey}")
    private String SECRET_KEY;
    
    @Inject
    private SQSUtils sqsUtils;

    @Inject
    private PushUtils pushUtils;
    
    private final Executor executor = Executors.newFixedThreadPool(9, new ThreadFactory() {
        private final AtomicInteger threadNumber = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "receiver-push-" + threadNumber.getAndIncrement());
            t.setDaemon(false);
            t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    });
 
  public void send() {

        List<PushContainer> list;
        do {
            list = sqsUtils.receive(Constants.SQS_RECEIVE_COUNT);

            if (CollectionUtils.isEmpty(list) == false) {

                final List<PushContainer> candidates = new ArrayList<>(list);
                FutureTask<Integer> task = new FutureTask<>(() -> {
                    AWSCredentials credentials = new BasicAWSCredentials(ACCESS_KEY, SECRET_KEY);
                    AmazonSNS snsClient = new AmazonSNSAsyncClient(credentials);
                    String threadId = Thread.currentThread().getName();
                    String endpoint = ENDPOINTS.get((Integer.parseInt(StringUtils.substring(threadId, threadId.length() - 1, threadId.length()))) % ENDPOINTS.size());

                    if (StringUtils.isNotEmpty(endpoint)) {
                        snsClient.setEndpoint(endpoint);
                        log.info("endpoint: {}", endpoint);
                    } else {
                        snsClient.setEndpoint(SNS_ENDPOINT);
                        log.info("endpoint: default value: {}", SNS_ENDPOINT);
                    }

                    int successCount = 0;
                    for (PushContainer pc : candidates) {
                        boolean result = pushUtils.sendMessage(snsClient, pc.getDevice(), pc.getMessage());
                        if (result) {
                            log.info("[SUCCESS] send message succeeded...userId: [{}], pushToken: [{}]", pc.getDevice().getUserId(), pc.getDevice().getPushToken());
                            successCount++;
                        } else {
                            log.info("[FAIL] send message failed...userId: [{}], pushToken: [{}]", pc.getDevice().getUserId(), pc.getDevice().getPushToken());
                        }
                    }
                    return successCount;
                });
                executor.execute(task);
            }
        } while (CollectionUtils.isEmpty(list) == false);
    }
}