Giới thiệu về Apache Camel

Apache Camel cũng là một trong những open-source hỗ trợ chúng ta hiện thực các ứng dụng Enterprise Service Bus bên cạnh Spring Integration, … Nó hỗ trợ hầu hết các pattern được đề cập trong Enterprise Integration Pattern .Trong bài viết, chúng ta sẽ cùng nhau tìm hiểu những khái niệm cơ bản của Apache Camel và làm một ví dụ để xem Apache Camel hoạt động như thế nào các bạn nhé!

Cơ bản về Apache Camel

Apache Camel sử dụng khái niệm Route thay vì Message Channel của Spring Integration để integrate, trao đổi data giữa các hệ thống, endpoint khác nhau. Các bạn có thể định nghĩa phần integration sử dụng Route này của Apache Camel sử dụng Domain Specific Language (DSL).

Một số khái niệm cơ bản mà các bạn cần biết khi làm việc với Apache Camel đó là:

Message

Message chứa data mà chúng ta muốn process, được truyền từ hệ thống này qua hệ thống khác sử dụng Route. Một Message sẽ bao gồm thông tin các headers, body chứa data mà chúng ta đang process, message ID để identify với các Message khác, và timestamp lúc Message được khởi tạo. Interface Message của Apache Camel sẽ đảm nhận việc định nghĩa các thông tin này:

Exchange

Exchange là một đối tượng chứa thông tin Message được khởi tạo khi một Message được gửi tới một Consumer. Nó sẽ xác định việc trao đổi giữa 2 hệ thống khác nhau là one-way hay là request-response.

Endpoint

Endpoint là các hệ thống sẽ nhận, process và gửi Message trong một Route của ứng dụng Apache Camel.

Component

Component là tập hợp các công cụ có sẵn của Apache Camel, giúp chúng ta dễ dàng tương tác với các hệ thống bên ngoài khác, process hoặc transform data mà không cần phải viết nhiều code. Ví dụ như các bạn thể sử dụng Component JDBC để làm việc với database chẳng hạn: các bạn không cần phải viết code mở kết nối đến database, chỉ cần truyền câu query để lấy được kết quả.

Các bạn có thể tham khảo thêm ở https://camel.apache.org/components/next/index.html để xem danh sách Component mà Apache Camel đã hỗ trợ nhé.

Processor

Processor là một interface giúp chúng ta có thể thêm các xử lý business logic bằng code Java. Nó có một phương thức là process() với tham số là đối tượng Exchange mà mình đã đề cập ở trên, giúp các bạn có thể lấy thông tin về Message đang được Route đi và thêm code Java để process Message theo ý của mình.

Một khái niệm nữa mà mình cũng muốn đề cập ở đây với các bạn là về interface CamelContext. Đây là đối tượng giúp chúng ta quản lý thông tin của một ứng dụng Apache Camel bao gồm thông tin các Route, các Component mà chúng ta sử dụng, các Endpoint, … Chúng ta có thể thêm, xoá Route, Component nếu muốn, …

Ứng dụng ví dụ

Để có cái nhìn sơ qua về cách mà Apache Camel sẽ làm việc như thế nào, mình sẽ làm một ứng dụng cơ bản, cứ sau 5s sẽ gửi một Message với body là dòng chữ “Hello Khanh from Huong Dan Java” từ Endpoint này sang Endpoint khác để Endpoint khác đó in ra dòng chữ này các bạn nhé!

Mình sẽ tạo mới một Maven project:

với Apache Camel dependency như sau:

<dependency>
  <groupId>org.apache.camel</groupId>
  <artifactId>camel-core</artifactId>
  <version>3.20.2</version>
</dependency>

<groupId>org.apache.camel</groupId>

<artifactId>camel-core</artifactId>

</dependency>

Mình sẽ sử dụng Java DSL để định nghĩa Route trong bài viết này các bạn nhé!

Mình sẽ tạo mới đối tượng CamelContext trước:

CamelContext camelContext = new DefaultCamelContext();

1	CamelContext camelContext = new DefaultCamelContext();

Để định nghĩa một Route, các bạn có thể sử dụng class RouteBuilder của Apache Camel. Chúng ta cần implement phương thức configure() để định nghĩa Route như sau:

RouteBuilder routeBuilder = new RouteBuilder() {

  @Override
  public void configure() throws Exception {

  }
};

RouteBuilder routeBuilder = new RouteBuilder() {

@Override

public void configure() throws Exception {

}

};

Trong ví dụ của mình, thì để mỗi một 5s, một Message mới được tạo ra thì mình sẽ sử dụng component Scheduler và component Log. Các bạn để ý là component Log này chỉ support cho Log4j, Logback và Java Util Logging thôi nhé!

Mình sẽ sử dụng Log4J cho ví dụ của mình với Slf4j dành cho Log4J dependency như sau:

<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-api</artifactId>
  <version>2.0.6</version>
</dependency>
<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-log4j12</artifactId>
  <version>2.0.6</version>
</dependency>

</dependency>

</dependency>

Nội dung tập tin cấu hình cho Log4J log4j.properties trong thư mục src/main/resources như sau:

log4j.rootLogger=INFO, A1

log4j.appender.A1=org.apache.log4j.ConsoleAppender
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-4r %-5p [%t] %37c %3x - %m%n

log4j.rootLogger=INFO, A1

log4j.appender.A1=org.apache.log4j.ConsoleAppender

log4j.appender.A1.layout=org.apache.log4j.PatternLayout

log4j.appender.A1.layout.ConversionPattern=%-4r %-5p [%t] %37c %3x - %m%n

Cứ mỗi 5s khi Scheduler được trigger mình sẽ set data cho phần Body của Message là dòng chữ “Hello Khanh from Huong Dan Java”, sau đó thì log ra như sau:

RouteBuilder routeBuilder = new RouteBuilder() {

  @Override
  public void configure() throws Exception {
    // @formatter:off
    from("scheduler://hdj?delay=5000")
        .setBody(simple("Hello Khanh from Huong Dan Java"))
        .to("log:com.huongdanjava.apachecamel?showAll=true&multiline=true");
    // @formatter:on
  }
};

RouteBuilder routeBuilder = new RouteBuilder() {

@Override

public void configure() throws Exception {

// @formatter:off

from("scheduler://hdj?delay=5000")

.setBody(simple("Hello Khanh from Huong Dan Java"))

.to("log:com.huongdanjava.apachecamel?showAll=true&multiline=true");

// @formatter:on

}

};

Chúng ta sử dụng phương thức from() của class RouteBuilder để khởi nguồn cho Message được tạo ra và phương thức to() để route Message. Tham số của phương thức from() và cả phương thức to() được định dạng với cấu trúc là một chuỗi bao gồm tên Component mà các bạn muốn sử dụng cùng với các cấu hình liên quan đến Component này. Tên component và các cấu hình sẽ phân biệt bằng dấu 2 chấm ở giữa.

Trong ví dụ trên thì mình sử dụng component Scheduler với cấu hình liên quan đến tên của nó và thời gian delay, còn component Log thì mình cấu hình về category và cách log message được hiển thị.

Phương thức setBody() mình sử dụng ở trên dùng để gán giá trị cho phần Body của Message.

Sau khi đã định nghĩa Router xong, chúng ta cần thêm nó vào CamelContext rồi start CamelContext lên như sau:

camelContext.addRoutes(routeBuilder);

camelContext.start();
Thread.sleep(100000);
camelContext.stop();

camelContext.addRoutes(routeBuilder);

camelContext.start();

Thread.sleep(100000);

camelContext.stop();

Vì mình đang chạy ứng dụng kiểu Java Console nên mình cũng cần Thread.sleep() để chương trình đủ thời gian để chạy. Và sau khi Thread.sleep() xong thì chúng ta có thể stop CamelContext để stop ứng dụng.

Toàn bộ code của ứng dụng ví dụ của mình như sau:

package com.huongdanjava.apachecamel;

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

public class Main {

  public static void main(String[] args) throws Exception {
    try (CamelContext camelContext = new DefaultCamelContext();) {

      RouteBuilder routeBuilder = new RouteBuilder() {

        @Override
        public void configure() throws Exception {
          // @formatter:off
          from("scheduler://hdj?delay=5000")
              .setBody(simple("Hello Khanh from Huong Dan Java"))
              .to("log:com.huongdanjava.apachecamel?showAll=true&multiline=true");
          // @formatter:on
        }
      };
      camelContext.addRoutes(routeBuilder);

      camelContext.start();
      Thread.sleep(100000);
      camelContext.stop();
    }
  }

}

package com.huongdanjava.apachecamel;

import org.apache.camel.CamelContext;

import org.apache.camel.builder.RouteBuilder;

import org.apache.camel.impl.DefaultCamelContext;

public class Main {

public static void main(String[] args) throws Exception {

try (CamelContext camelContext = new DefaultCamelContext();) {

RouteBuilder routeBuilder = new RouteBuilder() {

@Override

public void configure() throws Exception {

// @formatter:off

from("scheduler://hdj?delay=5000")

.setBody(simple("Hello Khanh from Huong Dan Java"))

.to("log:com.huongdanjava.apachecamel?showAll=true&multiline=true");

// @formatter:on

}

};

camelContext.addRoutes(routeBuilder);

camelContext.start();

Thread.sleep(100000);

camelContext.stop();

}

Kết quả khi mình chạy ví dụ này như sau:

Các bạn sẽ thấy sau mỗi 5s, ứng dụng sẽ trigger tạo mới một Message, sau đó thì in nội dung của Message này ra.

Khanh Nguyen

Share this post

Cơ bản về Apache Camel

Message

Exchange

Endpoint

Component

Processor

Ứng dụng ví dụ

Add Comment Hủy