Apache Kafka + Apache Camel

บทความนี้จะนำเสนอการ ติดตั้ง Apache Kafka บน Mac Os และ produce & consumer message จาก Apache Kafka ด้วย Apache ActiveMQ กันนะครับ

1. Installation Kafka server on OS X via Homebrew.

$ brew search kafka
$ brew install kafka

2. Start Zookeeper ก่อนที่จะ start Kafka.

$ zkserver start

3. tart the Kafka server.

$ cd /usr/local/Cellar/kafka/0.10.1.0/bin
$ kafka-server-start.sh /usr/local/etc/kafka/server.properties

เป็นอันเสร็จเรียบร้อยสำหรับการติดตั้ง Kafka Server ซึ่ง มันจะรันอยู่ที่ port 9092 นะครับ ทดสอบ Telnet ดูได้ครับ

$ telnet localhost 9092
Trying ::1…
Connected to localhost.
Escape character is ‘^]’.

ต่อไปเรามาสร้าง Project Java ด้วย Maven กันครับ

Get Chiwa Kantawong (Pea)’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>kafka-camel</groupId>
    <artifactId>kafka-camel</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.4.0.RELEASE</version>
        <relativePath/>
    </parent>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
            <version>1.4.0.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring-boot</artifactId>
            <version>2.15.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-core</artifactId>
            <version>2.19.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-kafka</artifactId>
            <version>2.19.2</version>
        </dependency>
    </dependencies>

</project>

application.yml

kafka:
    server : localhost
    port : 9092
    topic : my-topic
    channel : my-chanel

สร้าง Configuration สำหรับ Camel

package com.zengcode.camel.kafka.configuration;

import org.apache.camel.CamelContext;
import org.apache.camel.ThreadPoolRejectedPolicy;
import org.apache.camel.spi.ThreadPoolProfile;
import org.apache.camel.spring.boot.CamelContextConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MainConfiguration {

    @Bean
    CamelContextConfiguration contextConfiguration() {

        return new CamelContextConfiguration() {

            @Override
            public void beforeApplicationStart(CamelContext context) {
                // your custom configuration goes here
                ThreadPoolProfile threadPoolProfile = new ThreadPoolProfile();
                threadPoolProfile.setId("MyDefault");
                threadPoolProfile.setPoolSize(10);
                threadPoolProfile.setMaxPoolSize(15);
                threadPoolProfile.setMaxQueueSize(250);
                threadPoolProfile.setKeepAliveTime(25L);
                threadPoolProfile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort);
                context.getExecutorServiceManager().registerThreadPoolProfile(threadPoolProfile);

            }
        };
    }
}

สร้าง Camel Router

package com.zengcode.camel.kafka.router;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
public class MainRouter extends RouteBuilder {

    @Autowired
    private ProducerTemplate producerTemplate;

    @Override
    public void configure() throws Exception {

        from("kafka:{{kafka.topic}}?brokers={{kafka.server}}:{{kafka.port}}&groupId={{kafka.channel}}")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        System.out.println("Message Body : " + exchange.getIn().getBody());
                    }
                });

        from("timer://producer?period=1000")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        String message = UUID.randomUUID().toString();
                        producerTemplate.sendBody("kafka:{{kafka.topic}}?brokers={{kafka.server}}:{{kafka.port}}", message);
                    }
                });



    }
}

สร้าง Application.java

package com.zengcode.camel.kafka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class Application {

    public static void main(String[] args) throws Exception {
        SpringApplication.run(Application.class, args);
        while (true) {}
    }
}

แล้วก็ทดลองรันดูจะได้ ผลลัพท์ประมาณนี้ครับ

2017–08–16 13:34:36.663 INFO 55670 — — [imer://producer] o.a.kafka.common.utils.AppInfoParser : Kafka version : 0.10.2.0
2017–08–16 13:34:36.664 INFO 55670 — — [imer://producer] o.a.kafka.common.utils.AppInfoParser : Kafka commitId : 576d93a8dc0cf421
2017–08–16 13:34:38.344 INFO 55670 — — [sumer[my-topic]] o.a.k.c.c.internals.AbstractCoordinator : Successfully joined group my-chanel with generation 4
2017–08–16 13:34:38.344 INFO 55670 — — [sumer[my-topic]] o.a.k.c.c.internals.ConsumerCoordinator : Setting newly assigned partitions [] for group my-chanel

Message Body : 80859ab9–658e-4394–9b12–256a998e40d2
Message Body : 35d2da3a-16e2–4efa-8283–5c9c0f19f294
Message Body : 9f15946e-a24c-4312–9692–4024e6717bd4
Message Body : ed220406-b0fa-4655–91d6–6a04ce8777ee
Message Body : 244e29ec-4d1f-49b3-ac42–1544dda79cb3
Message Body : 9653bfcf-0d7f-494c-a7a1-b7defedac901
Message Body : 448ebdc7–591a-4e23–9109-c1e8efdb7a0c
Message Body : 26c9cd58–9a22–42fb-9df1–0a486376fbc5
Message Body : 124d25ae-5f41–4253–8ece-52b47b654fb0

NOTE : create, delete and alter number of partition

kafka-topics — create — zookeeper localhost:2181 — replication-factor 1 — partitions 10 — topic my-topic

kafka-topics — alter — zookeeper localhost:2181 — partitions 10 — topic my-topic

kafka-topics — delete — zookeeper localhost:2181 — topic my-topic