Home > Java > JMS(ActiveMQ)による非同期処理の実装

ActiveMQを使ったJMSのメール送信サンプルを紹介します。

ActiveMQは下記からダウンロードが可能です。

AcitveMQ:
http://activemq.apache.org/download.html

また、サンプルでは別途下記も必要になります。

Commons Email:
Commons Email Downloads

JavaMail:
http://java.sun.com/products/javamail/

JAF:
http://java.sun.com/products/javabeans/jaf/downloads/index.html

まずは普通にmail送信を行った場合。

test パッケージにTestNonJms.java として作成します。

PLAIN TEXT
JAVA:
  1. package test;
  2.  
  3. import org.apache.commons.mail.EmailException;
  4. import org.apache.commons.mail.SimpleEmail;
  5.  
  6. public class TestNonJms {
  7.  
  8. public static void main(String[] args){
  9. final long startTime = System.currentTimeMillis();
  10.  
  11. try {
  12. final SimpleEmail email = new SimpleEmail();
  13. email.setFrom("送信者アドレス");
  14. email.addTo("受信者アドレス");
  15. email.setSubject("JMS test");
  16. email.setContent("メールのテストです。", "text/plain; charset=ISO-2022-JP");
  17. email.setHostName("SMTPサーバーアドレス");
  18. email.send();
  19. } catch (EmailException e) {
  20. e.printStackTrace();
  21. }
  22.  
  23. final long endTime = System.currentTimeMillis();
  24.  
  25. System.out.println("non Jms:" + (endTime - startTime));
  26. }
  27. }

メールサーバの性能、トラフィック等によって値は変動しますが、3回実行して下記の様になりました。

PLAIN TEXT
CODE:
  1. non Jms:1672
  2. non Jms:1625
  3. non Jms:1625

続いてJMSを使用した場合です。

test.serverパッケージにTestJmsServer.javaとしておきます。
MessageBrokerよりメッセージを取得し処理するクラスになります。

PLAIN TEXT
JAVA:
  1. package test.server;
  2.  
  3. import javax.jms.JMSException;
  4. import javax.jms.ObjectMessage;
  5. import javax.jms.Queue;
  6. import javax.jms.QueueConnection;
  7. import javax.jms.QueueReceiver;
  8. import javax.jms.QueueSession;
  9.  
  10. import org.apache.activemq.ActiveMQConnection;
  11. import org.apache.activemq.ActiveMQConnectionFactory;
  12. import org.apache.commons.mail.EmailException;
  13. import org.apache.commons.mail.SimpleEmail;
  14.  
  15. import test.TestMessage;
  16.  
  17. public class TestJmsServer {
  18. public static boolean bQuit = false;
  19.  
  20. public static void main(String[] args) {
  21. QueueConnection connection = null;
  22. QueueSession session = null;
  23. QueueReceiver receiver = null;
  24.  
  25. try{
  26. ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
  27. connection = factory.createQueueConnection();
  28.  
  29. session = connection.createQueueSession(false,QueueSession.AUTO_ACKNOWLEDGE);
  30. Queue queue = session.createQueue("TestMailQueue");
  31.  
  32. receiver = session.createReceiver(queue);
  33.  
  34. connection.start();
  35.  
  36. while(!bQuit) {
  37. ObjectMessage msg = (ObjectMessage)receiver.receive();
  38. TestMessage mailMessage = (TestMessage)msg.getObject();
  39.  
  40. try {
  41. final SimpleEmail email = new SimpleEmail();
  42. email.setFrom(mailMessage.getFromAddr());
  43. email.addTo(mailMessage.getToAddr());
  44. email.setSubject(mailMessage.getSubject());
  45. email.setContent(mailMessage.getMessage(), "text/plain; charset=ISO-2022-JP");
  46. email.setHostName("SMTPサーバーアドレス");
  47. email.send();
  48.  
  49. } catch (EmailException e) {
  50. e.printStackTrace();
  51. bQuit = true;
  52. }
  53. }
  54.  
  55. } catch (JMSException e) {
  56. e.printStackTrace();
  57.  
  58. } finally {
  59. try {
  60. if (receiver != null) {
  61. receiver.close();
  62. }
  63. if (session != null) {
  64. session.close();
  65. }
  66. if (connection != null) {
  67. connection.close();
  68. }
  69.  
  70. } catch (JMSException e) {
  71. e.printStackTrace();
  72. }
  73. }
  74. }
  75. }

testパッケージにTestMessage.javaを作成します。
メッセージを格納するクラスになります。

PLAIN TEXT
JAVA:
  1. package test;
  2.  
  3. import java.io.Serializable;
  4.  
  5. public class TestMessage implements Serializable {
  6.  
  7. private static final long serialVersionUID = 5997289309760126077L;
  8.  
  9. private String subject;
  10. private String fromAddr;
  11. private String toAddr;
  12. private String message;
  13.  
  14. public String getSubject() {
  15. return subject;
  16. }
  17. public void setSubject(String subject) {
  18. this.subject = subject;
  19. }
  20. public String getFromAddr() {
  21. return fromAddr;
  22. }
  23. public void setFromAddr(String fromAddr) {
  24. this.fromAddr = fromAddr;
  25. }
  26. public String getToAddr() {
  27. return toAddr;
  28. }
  29. public void setToAddr(String toAddr) {
  30. this.toAddr = toAddr;
  31. }
  32. public String getMessage() {
  33. return message;
  34. }
  35. public void setMessage(String message) {
  36. this.message = message;
  37. }
  38. }

testパッケージにTestJms.javaを作成します。
JMSを利用してTestNonJmsと同等の処理を行うクラスになります。

PLAIN TEXT
JAVA:
  1. package test;
  2.  
  3. import javax.jms.JMSException;
  4. import javax.jms.Message;
  5. import javax.jms.Queue;
  6. import javax.jms.QueueConnection;
  7. import javax.jms.QueueConnectionFactory;
  8. import javax.jms.QueueSender;
  9. import javax.jms.QueueSession;
  10.  
  11. import org.apache.activemq.ActiveMQConnection;
  12. import org.apache.activemq.ActiveMQConnectionFactory;
  13.  
  14. public class TestJms {
  15.  
  16. public static void main(String[] args){
  17.  
  18. long startTime = System.currentTimeMillis();
  19.  
  20. QueueConnection connection = null;
  21. QueueSession session = null;
  22. QueueSender sender = null;
  23.  
  24. try {
  25.  
  26. QueueConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
  27. connection = factory.createQueueConnection();
  28.  
  29. session = connection.createQueueSession(false,QueueSession.AUTO_ACKNOWLEDGE);
  30. Queue queue = session.createQueue("TestMailQueue");
  31. sender= session.createSender(queue);
  32.  
  33. connection.start();
  34.  
  35. TestMessage mailMessage = new TestMessage();
  36. mailMessage.setSubject("ActiveMQ test");
  37. mailMessage.setFromAddr("送信者アドレス");
  38. mailMessage.setToAddr("受信者アドレス");
  39. mailMessage.setMessage("メールのテストです。");
  40.  
  41. Message msg = session.createObjectMessage(mailMessage);
  42.  
  43. sender.send(msg);
  44.  
  45. } catch (JMSException e) {
  46. e.printStackTrace();
  47.  
  48. } finally {
  49. try {
  50. if (sender != null) {
  51. sender.close();
  52. }
  53. if (session != null) {
  54. session.close();
  55. }
  56. if (connection != null) {
  57. connection.close();
  58. }
  59. } catch (JMSException e) {
  60. e.printStackTrace();
  61. }
  62. }
  63.  
  64. long endTime = System.currentTimeMillis();
  65.  
  66. System.out.println( "TestJms:" + (endTime - startTime) );
  67. }
  68. }

まず、コマンドプロンプトよりActiveMQのアーカイブに含まれているbin/activemq.bat（MessageBroker）を実行しておきます。
続いてTestJmsServerを実行します。これでJMSが使える状態になりました。

TestJmsを実行してみます。
3回実行して下記の様になりました。

PLAIN TEXT
CODE:
  1. TestJms:687
  2. TestJms:671
  3. TestJms:656

TestNonJmsを実行した時よりも大分処理時間が短くなっている事が判ります。

この様にメールの送信や処理に多少の遅延が許させる処理等を非同期化する事でアプリケーションのパフォーマンスを上げる事が出来ます。