Pluggableな非同期処理

@JsonIgnoreProperties(ignoreUnknown = true)
public class JsonBase {
	
	private String serviceId;

	public String getServiceId() {
		return serviceId;
	}

	public void setServiceId(String serviceId) {
		this.serviceId = serviceId;
	}

}

public class MailJson extends JsonBase {

	private Data data;

	public Data getData() {
		return data;
	}

	public void setData(Data data) {
		this.data = data;
	}

	public class Data {
		private String subject; // メールのタイトル
		private String to; // メールの宛先
		private String from; // メールの送信元
		private String body; // メールの本文

		public String getSubject() {
			return subject;
		}

		public void setSubject(String subject) {
			this.subject = subject;
		}

		public String getTo() {
			return to;
		}

		public void setTo(String to) {
			this.to = to;
		}

		public String getFrom() {
			return from;
		}

		public void setFrom(String from) {
			this.from = from;
		}

		public String getBody() {
			return body;
		}

		public void setBody(String body) {
			this.body = body;
		}

	}
}

public interface Service {
	
	/**
	 * メッセージを処理するメソッドで、引数にはメッセージの内容(JSON)を指定する
	 * @param message メッセージ
	 */
	public void exec(String message);
}

public class SendMailService implements Service {

	private static ObjectMapper objectMapper = new ObjectMapper();
	
	@Override
	public void exec(String message) {
		try {
			// ここでは、JSONの内容を標準出力に出しているだけだが、実際にはメール送信処理等、ロジックを記載する
			MailJson result = objectMapper.readValue(message,MailJson.class);
			System.out.println("ServiceId:" + result.getServiceId());
			System.out.println("Subject:" + result.getData().getSubject());
			System.out.println("From:" + result.getData().getFrom());
			System.out.println("To:" + result.getData().getTo());
			System.out.println("Body:" + result.getData().getBody());
		} catch (IOException e) {
			e.printStackTrace();
		}		
	}
}

@Configuration
public class WorkerServiceConfig {

	// Bean名には、メッセージのJSONの「serviceId」にある値と同じものを指定する
	@Bean(name = "SendMailService")
	public Service getSendMailService() {

		return new SendMailService();

	}
}

■ AzureBillingManagerWorkerApplication.java

Workerプロセスのメインクラスになります。ここから先の処理はSpring Boot本家のホームページの以下のURLを参考にしています。詳しくは以下をご参照下さい。

https://spring.io/guides/gs/async-method/

AsyncConfigurerSupportを継承し、getAsyncExecutor()でスレッド処理の定義をしています。

@SpringBootApplication
@EnableAsync
@Import(WorkerServiceConfig.class)
public class AzureBillingManagerWorkerApplication extends AsyncConfigurerSupport {

	public static void main(String[] args) {
		SpringApplication.run(AzureBillingManagerWorkerApplication.class, args);
	}

	@Override
	public Executor getAsyncExecutor() {
		// スレッドプールの設定を記載する
		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
		executor.setCorePoolSize(2);
		executor.setMaxPoolSize(2);
		executor.setQueueCapacity(500);
		executor.setThreadNamePrefix("Thread-");
		executor.initialize();
		return executor;
	}
}

■ AppRunner.java

AzureBillingManagerWorkerApplication.javaから呼ばれて、スレッドを生成するクラスです。

@Component
public class AppRunner implements CommandLineRunner {

	private final RetrieveQueue retrieveQueue;

	public AppRunner(RetrieveQueue retrieveQueue) {
		this.retrieveQueue = retrieveQueue;
	}

	@Override
	public void run(String... arg0) throws Exception {
		// 2つのスレッドを生成している
		CompletableFuture<String> f1 = retrieveQueue.exec();
		CompletableFuture<String> f2 = retrieveQueue.exec();
		
		// スレッドが全て終了するまでメイン処理を終了しない
		while (!(f1.isDone() && f2.isDone())) {
			Thread.sleep(10);
		}
	}
}

■ RetrieveQueue.java

キューを監視し、キューにメッセージ(JSON)が入ってきたら、JSONのserviceIdに対応するBeanをApplication Contextから取得し、実行するクラスです。今回非同期で処理させたいのはこのクラスになります。非同期で処理したいメソッドに@Asyncをつけるのがポイントになります。

@Service
public class RetrieveQueue {

	private static final Logger logger = LoggerFactory.getLogger(RetrieveQueue.class);
	private static final String connection = "XXXXXX";
	private static final String queueName = "mail-sender";
	private static ObjectMapper objectMapper = new ObjectMapper();

	@Autowired
	private ApplicationContext applicationContext;

	@Async
	public CompletableFuture<String> exec() {

		try {
			// CONNECTION STRINGの情報をもとにキューの入れ物に接続する
			CloudStorageAccount storageAccount = CloudStorageAccount.parse(connection);

			// キューのクライアントを生成する
			CloudQueueClient queueClient = storageAccount.createCloudQueueClient();

			// キュー名をセットする
			CloudQueue queue = queueClient.getQueueReference(queueName);

			// 無限ループによりキューを監視する
			while (true) {

				// キューからメッセージを取得する
				CloudQueueMessage retrievedMessage = queue.retrieveMessage();

				if (retrievedMessage != null) {
					// JSONからサービスIDを取得して、Javaオブジェクトにマップする
					JsonBase result = objectMapper.readValue(retrievedMessage.getMessageContentAsString(),
							JsonBase.class);
					
					// サービスIDに対応したBeanをApplication Contextから取得する
					com.sios.azure.billing.worker.service.Service service = (com.sios.azure.billing.worker.service.Service) applicationContext
							.getBean(result.getServiceId());
					
					// サービスに定義された処理を実行する
					service.exec(retrievedMessage.getMessageContentAsString());
					
					// キューを削除する。
					queue.deleteMessage(retrievedMessage);
				}
			}
		} catch (Exception e) {
			logger.error("Exception occurred while thread is running...", e);
		}

		return CompletableFuture.completedFuture("Thread is done!!");
	}
}