مقدمه ای بر کَمِل

در قسمت قبل الگوهای ادغام سازمانی (Enterprise Integration Patterns) را به صورت مختصر معرفی کردیم. در ادامه با یکی از ابزارهایی که برای پیاده سازی این الگوها استفاده می شود، آشنا می شویم.

کمل (Apache Camel)، یکی از چهارچوب هایی است که امکان ادغام سرویس های تولید کننده و مصرف کننده را فراهم می کند. این پروژه در اوایل سال 2007 آغاز شده و در حال حاضر به صورت منبع باز و تحت مجوز Apache 2.0 در دسترس قرار گرفته است.

جهت بررسی بیشتر، پیشنهاد می کنیم به Camel in Action, Second Edition مراجعه کنید.

مقدمه

به صورت کلی کمل این امکان را برای شما فراهم می کند که برای سیستم پیامگرای خود، قواعد مسیریابی تعریف کنید. این قواعد را می توان به صورت زیر بیان کرد اینکه پیام ها:

• از کدام منبع دریافت شوند
• به چه صورت پردازش شوند
• به کدام مقصد هدایت شوند

همانطور که قبلا هم اشاره کردیم، از چالش های مهم ادغام سرویس ها، پیاده سازی مختلف آن ها است. هر سرویس انتظار دارد که نوع داده خاصی را در پیام ها استخراج کند. همچنین ممکن است از پروتکل های مختلفی پیشتیبانی بکند. کمل با ارائه واسط های مختلف و پیاده سازی حالت های بسیار از آن ها، تا حد ممکن این مشکلات را از توسعه دهنده پنهان می کند.

برخی از مهمترین امکاناتی که کمل در اختیار ما قرار می دهد عبارتند از:

1. واسط مسیریابی

هسته اصلی کمل که امکان انتقال پیام ها را میان اجزاء مختلف با توجه به مسیر های ایجاد شده به وسیله توسعه دهنده فراهم می کند. این مسیرها به وسیله الگوهای ادغام سازمانی (EIP) و زبان واسط (DSL) تنظیم می شوند که در ادامه شرح خواهیم داد.

2. کامپوننت های گسترده

در کمل، کتاب خانه گسترده ای از کامپوننت های مختلف ارائه شده که هر کدام برای بهره وری از واسط ها و تبدیل داده استفاده می شوند. کامپوننت ها امکان تعامل با سایر نرم افزار ها را فراهم می کنند.

3. مطابق با الگوهای ادغام سازمانی

با وجود اینکه الگوهای ادغام سازمانی EIP مشکلات ادغام و راه حل آنها را توصیف می کند؛ اما آن ها را به صورت رسمی پیاده سازی نمی کند. کمل با ارائه زبانی خاص و مشترک میان همه توسعه دهندگان تلاش کرده تا حد ممکن این شکاف را برطرف کند.

4. زبان واسط domain-specific language (DSL)

کمل با ارائه زبان واسط تلاش می کند یکپارچگی خاصی در زمینه پیاده سازی الگوهای ادغام سازمانی ارائه بدهد و توسعه دهنده به جای تمرکز بر پیاده سازی این الگوها، بر ارائه راه حل تمرکز کند. این زبان واسط به شکل کد جاوا یا فرمت XML قابل استفاده است:

به منظور سادگی در ادامه از Java DSL استفاده خواهیم کرد. برای شروع پیشنهاد می کنیم مثال کپی کردن فایل از مبدا به مقصدی خاص را از گیتهاب بررسی کنید.

معماری کمل

در ادامه هر یک از اجزاء معماری کمل را به صورت مختصر بررسی می کنیم

Camel context

همه اجزا را کنار هم قرار داده و امکان اجرای کمل را فراهم می کند. هر مسیر(Route)ی که به وسیله توسعه دهنده تنظیم می شود، به وسیله Camel context نگهداری شده و به بهره برداری می رسد.

Routing engine

همان چیزی که پیام ها را میان اجزاء تولید کننده و مصرف کننده انتقال می دهد. Routing engine استفاده ای برای توسعه دهنده ندارد؛ اما خوب است بدانیم که انتقال پیام ها به وسیله آن انجام می شود.

Routes

مسیرها به وسیله توسعه دهنده تعریف می شوند. یک مسیر ساده می تواند زنجیره ای از پردازش ها را میان مبدا و مقصد داشته باشد که به وسیله Processor ها انجام می شود. در کمل هر مسیر فقط یک مبدا خواهد داشت که تولید کننده و مصرف کنندگان را از یکدیگر مستقل می کند؛ بنابراین حتی اگر به شکل زیر چندین مبدا برای یک مسیر در نظر بگیریم:

 from('jms:queue:A', 'jms:queue:B', 'jms:queue:C').to('jms:queue:D');

در نهایت کمل آن را به شکل زیر برداشت خواهد کرد:

from('jms:queue:A').to('jms:queue:D'); 
from('jms:queue:B').to('jms:queue:D'); 
from('jms:queue:C').to('jms:queue:D');

Processors

از مهمترین اجزاء کمل که امکان دسترسی به پیام ها و مدیریت آن ها را فراهم می کند. Processor ها می توانند همزمان به صورت تولید کننده و مصرف کننده عمل کنند؛ به گونه ای که هر Processor پیام های پردازش شده به وسیله Processor قبلی خودش را دریافت کند. مثال قسمت قبل را در نظر بگیرید که بخواهیم پیام های دریافت شده از طرف کاربر را ابتدا رمزگشایی کرده و پس از انجام عملیات احراز هویت پیام های اسپم را نادیده بگیریم. این ویژگی به شکل زیر به وسیله Java DSL قابل پیاده سازی می باشد:

 from('direct:start')
.process(decryptionProcessor)
.process(authenticationProcessor)
.process(spamTrashProcessor)
.to('direct:end');

نحوه پیاده سازی Processor ها در ادامه بررسی خواهیم کرد.

Components

کامپوننت ها امکان تعامل با سایر واسط ها را با وجود پروتکل های گوناگون و فرمت داده متفاوت فراهم می کنند. با استفاده از آن ها می توانید پیام های موجود در یک کارگزار پیام (Message broker) را استخراج کنید یا به یک دیتابیس خاص کوئری بزنید یا حتی رویداد های یک ربات تلگرام را پردازش کرده و نتیجه را برگردانید. به طور خاص کامپوننت ها به وسیله نام خود در URI مورد استفاده قرار می گیرند. به عنوان مثال فرض کنید می خواهیم به فایلی در مسیر خاص دسترسی پیدا کنیم:

from('file:data/inbox')

در این حالت اسم file، در حقیقت FileComponent را مشخص می کند که امکان دسترسی به مسیر data/inbox را فراهم می کند. در کمل کامپوننت های متنوعی تعریف شده اما امکان توسعه کامپوننت جدید هم وجود دارد. کامپوننت ها به عنوان کارخانه ای از Endpoint ها عمل می کنند.

Endpoints

به عنوان واسطی میان اجزا و سیستم پیامگرا مورد استفاده قرار می گیرند. تولید کننده و مصرف کننده به وسیله Endpoint، پیام ها را دریافت کرده یا ارسال می کنند.

در کمل، پیام ها در حقیقت به صورت یک Exchange میان اجزاء رد و بدل می شوند که به وسیله Endpoint ها مدیریت می شوند.

در تصویر فوق فیلد های یک Exchange را مشاهده می کنید. یکی از مهمترین آنها MEP است. یک الگوی تبادل پیام (Message Exchange Pattern) یا به اختصار MEP، نوع پیامگرایی را به صورت یک طرفه یا درخواست پاسخ مشخص می کند. MEP موجود در یک Exchange، مقادیر زیر را میتواند داشته باشد:

1. InOnly

پیام گرایی یک طرفه. ساده ترین مثال آن می تواند استفاده از یک کارگزار پیام باشد.

2. InOut

مدل درخواست پاسخ که به عنوان مثال می تواند در HTTP مورد استفاده قرار بگیرد.

در مثال فوق می بینید که Exchange هایی که فیلد MEP برابر با InOut دارند، به مصرف کننده بازگردانده می شوند.

Routing

در این قسمت نوشتن مسیر در کمل را بررسی خواهیم کرد و بخشی از الگوهای EIP را به وسیله آن پیاده سازی می کنیم.

امروزه مسیریابی را می توان در بسیاری از روزمرگی های خودمان مشاهده کنیم. به عنوان مثال، وقتی نامه ای را پست می کنید؛ ممکن است قبل از رسیدن به آدرس نهایی، از چندین شهر عبور کند. ایمیلی که ارسال می کنید؛ قبل از رسیدن به مقصد نهایی از طریق بسیاری از سیستم های شبکه کامپیوتری هدایت می شود. در همه موارد، مسیریاب انتخاب می کند که بسته دریافتی را از چه راهی به مقصد بفرستد.

کمل با بهره گیری از الگوهای تعریف شده معماری پیامگرا، امکان توسعه سیستم های مبتنی بر این رویکرد را فراهم می کند. در ادامه مثال هایی از پیاده سازی موارد ساده به وسیله Java DSL را بررسی خواهیم کرد.

استفاده از تایمر

به عنوان اولین مثال، نشان می دهیم که چگونه پیام خاصی را طی یک بازه زمانی مشخص در خروجی چاپ کنیم.

برای تعریف مسیر به وسیله Java DSL لازم است که از کلاس RouteBuilder ارث بری کنیم. این کلاس متد configure را در اختیار ما قرار می دهد که در آن پیاده سازی خودمان را انجام می دهیم:

public class MyRouteBuilder extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        from('timer:foo?period=5000')
            .transform(simple(&quotHello&quot))
            .log('${body}');
    }
}

کامپوننت Timer برای ایجاد پیام در یک بازه زمانی مشخص استفاده می شود. این پیام ها به صورت یک رویداد (event) تولید می شوند. در این مثال اسم کامپوننت را foo در نظر گرفته ایم و مقدار بازه زمانی دلخواه را برابر 5 ثانیه قرار داده ایم. در ادامه به وسیله متد transform، پیام Hello را در بدنه ورودی (Body) قرار داده و سپس آن را چاپ می کنیم.

همانطور که قبلا هم گفتیم، در کمل تمام مسیرهای تعریف شده به وسیله CamelContext نگهداری می شوند و به بهره برداری می رسند. لذا در متد Main:

public final class MyCamelApplication {
    public static void main(String[] args) throws Exception {
        CamelContext context = new DefaultCamelContext();
        context.addRoutes(new MyRouteBuilder());
        context.start();
    }
}

توجه داشته باشید که در مثال فوق کمل فرصتی برای چاپ پیام Hello نخواهد داشت زیرا ترد اصلی که به وسیله main ایجاد شده به انتها می رسد. برای حل این مشکل متد main دیگری ایجاد می کنیم:

import org.apache.camel.main.Main;

public class MyApplication {
    public static void main(String[] args) throws Exception {
        Main main = new Main(MyCamelApplication.class);
        main.run();
    }
}

که به خروجی زیر منجر می شود:

استفاده از Processor ها

کامپوننت Timer برای هر پیامی که ارسال می کند یک هدر HEADER_FIRED_TIME تنظیم می کند که زمان تولید آن پیام را به شکل روز هفته، ماه، ساعت و سال میلادی نشان می دهد. در این قسمت قصد داریم هدری به هر پیام اضافه کنیم؛ به طوری که روز هفته شمسی (شنبه، یکشنبه و ...) را بیان کند. این کار را به کمک یک پردازه (Processor) انجام می دهیم.

ابتدا یک Enum برای مدل کردن روزهای هفته در نظر میگیریم:

public enum WeekDays {
    
    Sat('شنبه'),
    Sun('یکشنبه'),
    Mon('دوشنبه'),
    Tue('سه شنبه'),
    Wed('چهارشنبه'),
    Thu('پنجشنبه'),
    Fri('جمعه');

    private final String name;

    WeekDays(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }
}

برای پیاده سازی یک پردازه، لازم است که از واسط org.apache.camel.Processor استفاده کنیم. این واسط متدی به نام process فراهم می کند که امکان دسترسی به Exchange دریافتی را می دهد:

public class TimerProcessor implements Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        final Message message = exchange.getMessage();
        final String firedTime = message.getHeader(
                TimerConstants.HEADER_FIRED_TIME, String.class
        );

        WeekDays weekDay = Arrays.stream(WeekDays.values())
                        .filter(weekDays -> firedTime.contains(weekDays.toString()))
                        .findFirst().get();

        exchange.getMessage().setHeader('dayOfWeek', weekDay.getName());
    }
}

در مثال فوق ابتدا پیام را از Exchange دریافتی استخراج کرده سپس با توجه به محتوای هدر پیام، مشخص می کنیم که کدام یک از روزهای هفته را باید در نظر بگیریم. در پایان نتیجه را به عنوان هدر جدیدی به نام dayOfWeek در پیام قرار می دهیم.

برای استفاده از TimerProcessor لازم است که آن را به Java DSL قبلی اضافه کنیم:

public class MyRouteBuilder extends RouteBuilder {

    @Override
    public void configure() throws Exception {
        from('timer:foo?period=5000')
            .transform(simple('Hello'))
            .process(new TimerProcessor())
            .log('${body} on ${header.dayOfWeek}');
    }
}

مشاهده می کنید که کمل امکان دسترسی به هدر و بدنه را در متد log فراهم می کند که منجر به خروجی زیر خواهد شد:

استفاده از Message Router

در قسمت الگوهای پیامگرایی دیدیم که مسیریاب پیام، فیلتری است که می تواند با استفاده از هدر های یک پیام، آن را به خروجی مناسب هدایت کند. در این بخش مثال قبل را توسعه می دهیم به گونه ای که پیام هایی که در روز زوج ساخته شده باشند به یک صف خاص، و روزهای فرد برای صف دیگر هدایت شود.

public class MyRouteBuilder extends RouteBuilder {

    private static final String DAY_OF_WEEK = 'dayOfWeek'
    private static final String EVEN_QUEUE = 'direct:even'
    private static final String ODD_QUEUE = 'direct:odd'
    private static final String ANOTHER_QUEUE = 'direct:another'


    @Override
    public void configure() throws Exception {
        from('timer:foo?period=5000')
            .transform(simple('Hello'))
            .process(new TimerProcessor())
            .choice()
                .when(exchange -> {
                    final String day = exchange.getMessage().getHeader(DAY_OF_WEEK, String.class);
                    return day.contains(WeekDays.Sat.getName()) ||
                           day.contains(WeekDays.Mon.getName()) ||
                           day.contains(WeekDays.Wed.getName());
                }).to(EVEN_QUEUE)
                .when(exchange -> {
                    final String day = exchange.getMessage().getHeader(DAY_OF_WEEK, String.class);
                    return day.contains(WeekDays.Sun.getName()) ||
                            day.contains(WeekDays.Tue.getName()) ||
                            day.contains(WeekDays.Thu.getName());
                }).to(ODD_QUEUE)
                .otherwise().to(ANOTHER_QUEUE)
            .end()
            .log('${body} on ${header.' + DAY_OF_WEEK + '}');
        
            // We have to process all queues in our messaging system
           from(EVEN_QUEUE).process();
           from(ODD_QUEUE).log('I\'ll be processed before logging body and header').process();
           from(ANOTHER_QUEUE).process();
    }
}

مشاهده می کنید که پیام ها با توجه به مقدار هدر DAY_OF_WEEK، به یکی از کامپوننت های direct:even، direct:odd یا direct:another هدایت می شوند. کامپوننت direct، امکان فراخوانی سنکرون برای پردازش پیام ها را فراهم می کند؛ به عبارت دیگر تا زمانی که پیام به وسیله زیر مجموعه های آن direct پردازش نشود؛ سراغ کامپوننت های دیگر نخواهد رفت. در ادامه پیام های دریافتی از هر یک از صف ها را به طور نمونه پردازش خواهیم کرد که منجر به خروجی زیر خواهد شد:

برخلاف direct، کامپوننت SEDA امکان فراخوانی آسنکرون را فراهم می کند؛ لذا با تغییر نام direct به seda خروجی به شکل زیر غیر قابل پیش بینی خواهد بود:

استفاده از Message broker

در ادامه پیام های تولید شده به وسیله تایمر را به یک کارگزار پیام ارسال می کنیم. در این مثال ما از ActiveMQ استفاده می کنیم؛ بنابراین وابستگی مربوط کامپوننت آن را به پروژه خود اضافه می کنیم:

<dependency>
  <groupId>org.apache.camel</groupId>
  <artifactId>camel-activemq</artifactId>
</dependency>

همچنین برای اتصال به ActiveMQ:

<dependency>  
    <groupId>org.apache.activemq</groupId>  
    <artifactId>activemq-all</artifactId>  
    <version>5.15.2</version> 
</dependency> 

حال ابتدا کامپوننت مورد نیاز را به Camel context اضافه می کنیم:

var connectionFactory = new ActiveMQConnectionFactory('vm://localhost');
var context = new DefaultCamelContext(); 
context.addComponent(
    'amq',    
    JmsComponent.jmsComponentAutoAcknowledge(connectionFactory)
); 

در اینجا برای اتصال به ActiveMQ از شی connectionFactory استفاده کرده ایم. ورودی vm://localhost امکان استفاده از یک Embedded broker را فراهم می کند. درنهایت پیام ها را به صف sample ارسال می کنیم:

from('timer:foo?period=1000')
.transform(simple('Hello'))
.to('amq:queue:sample');

پیوند های مفید

ایجاد پروژه اولیه کمل با توجه به وابستگی های مشخص شده: Kameleon

لیستی کامل از کامپوننت های کمل: Components

مثال هایی از پیاده سازی الگوهای سازمانی با استفاده از کمل: Camel-Enterprise-Integration-Patterns