sudo apt-get update -y sudo apt-get upgrade -y
sudo java -version
sudo add-apt-repository -y ppa:webupd8team/java
gpg: keyring `/tmp/tmpkjrm4mnm/secring.gpg' created gpg: keyring `/tmp/tmpkjrm4mnm/pubring.gpg' created gpg: requesting key EEA14886 from hkp server keyserver.ubuntu.com gpg: /tmp/tmpkjrm4mnm/trustdb.gpg: trustdb created gpg: key EEA14886: public key "Launchpad VLC" imported gpg: no ultimately trusted keys found gpg: Total number processed: 1 gpg: imported: 1 (RSA: 1) OK
sudo apt-get update
sudo apt-get install oracle-java8-installer -y
sudo java -version
java version "1.8.0_66" Java(TM) SE Runtime Environment (build 1.8.0_66-b17) Java HotSpot(TM) 64-Bit Server VM (build 25.66-b17, mixed mode)
پس از دریافت Apache Kafka، از طریق یک terminal جدید وارد پوشه Downloads شوید. سپس فایل مورد نظر را از حالت فشرده خارج کنید و وارد پوشه اصلی آن شوید. برای این کار از دستورات زیر استفاده کنید:
tar -xzf kafka_2.11-1.0.0.tgz cd kafka_2.11-1.0.0
سپس با استفاده از دستور "ls " لیست تمامی فایلها و فولدر‌های موجود در فولدر kafka_2.11-1.0.0 و را نمایش دهید:
در لیست نمایش داده شده دو فولد اصلی bin و config وجود دارند که به ترتیب فایلهای اجرایی و کانفیگهای پیشفرض و مورد نیاز، در آنها قرار دارند.
اجرای Apache ZooKeeper:
همانطور که در بخش قبل توضیح داده شد، Apache Kafka هیچ Stateی را در خود ذخیره و مدیریت نمی‌کند و اصطلاحا Stateless میباشد و مدیریت تمامی این Stateها را بر عهده Apache ZooKeeper قرار می‌دهد. بنابراین قبل از اینکه بخواهیم Kafka را اجرا کنیم، ابتدا باید Apache ZooKeeper را اجرا کنیم. برای اجرای Apache ZooKeeper در terminalی که قبلا باز کرده‌اید، دستور زیر را اجرا کنید:
bin/zookeeper-server-start.sh config/zookeeper.properties
با اجرای فایل zookeeper-server-start.sh و ارسال کانفیگ پیشفرضش در فولدر config با نام zookeeper.properties، قسمت مدیریت کننده Stateهای Kafka اجرا می‌شود.
اجرای Apache Kafka:
پس از اجرای Apache ZooKeeper، باید یک terminal جدید را باز کنید و با استفاده از دستور زیر kafka-server را با استفاده از کانفیگ پیشفرضش اجرا کنید:
bin/kafka-server-start.sh config/server.properties
به همین راحتی Kafka Server اجرا شده و آماده‌ی استفاده است. برای ادامه و نمایش دادن سایر قابلیتهای Kafka باید یک Topic جدید را ایجاد کنیم.
ایجاد Topic:
همانطور که می‌دانید تمامی پیامهای شما در Partitionهای Topic ذخیره می‌شوند؛ پس قبل از اینکه بخواهیم توسط یک Producer پیامی را ارسال کنیم یا اینکه بخواهیم توسط Consumer پیامی را دریافت کنیم، ابتدا باید Topic مربوطه را ایجاد کنیم. بهترین و عمومی‌ترین مثال برای نمایش قابلیتهای Publish-Subscribe در Kafka، مثال چت بین کاربران است که در آن یک کاربر به عنوان Producer عمل می‌کند و پیامهایی را برای سایر کاربران که نقش Consumer را دارند ارسال می‌کند. kafka-topics .sh در Kafka ابزاریست برای مدیریت Topic ها که با استفاده از آن می‌توانید Topicهای مورد نیاز را تعریف، ویرایش و یا حذف کنید.
یک terminal جدید را آغاز و با استفاده از دستور زیر یک Topic را با نام userChat ایجاد کنید:
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic userChat
با این دستور یک Topic در Kafka با نام userChat ایجاد می‌شود و مشخصات آن (منظور stateهای مرتبط با آن است)، در zookeeper نیز ثبت می‌شود.
توضیحات Optionها و آرگومانهایی که در دستور ایجاد Topic استفاده شدند به شرح زیر است:
create-- :
مشخص کننده این است که شما می‌خواهید یک Topic جدید را ایجاد کنید.
zookeeper localhost:2181--:
مشخص کننده آدرس سرور zookeeper است (سرور zookeeper بصورت پیشفرض از پورت 2181 استفاده می‌کند).
replication-factor 1--:
مشخص کننده این است که تنها یک کپی از این Topic در سرور ایجاد شود. البته در این مثال به دلیل اینکه تنها یک سرور از Kafka را اجرا کرده‌ایم در صورتیکه این مقدار را بیش از عدد 1 در نظر بگیریم، با خطا مواجه می‌شویم.
partitions 1--
تعداد Partitionهای این Topic را مشخص می‌کند. مقدار Partition در حالتیکه حتی یک سرور را نیز اجرا کرده‌ایم، می‌تواند بیش از 1 باشد؛ اما در این حالت Primary Broker برای تمام Partitionهای همین سرور در نظر گرفته می‌شود.
topic userChat--
نام Topic را مشخص می‌کند.
پس از اینکه Topic مورد نظر ایجاد شد، با استفاده از دستور زیر می‌توانیم لیست Topicهای ایجاد شده را مشاهده کنیم:
bin/kafka-topics.sh --list --zookeeper localhost:2181
توضیحات Optionها و آرگومانهایی که در دستور نمایش لیست Topicها استفاده شدند به شرح زیر است:
list--:
شما می‌خواهید لیست topicها را مشاهده کنید.
zookeeper localhost:2181--:
همانند دستور ایجاد، مشخص کننده سرور zookeeper میباشد.
تا اینجا برای ارسال و دریافت پیام در بین Applicationهای مختلف همه چیز آماده‌است. البته همانطور که در بخشهای بعدی این سری مقالات می‌بینیم، در عمل Producerها و Consumerها زیرسیستمهایی هستند که خود ما با استفاده از هر زبان برنامه نویسی پیاده سازی می‌کنیم. اما در این قسمت برای نمایش و تکمیل مثالمان، از ابزارهایی که خود Kafka در اختیار ما قرار داده استفاده می‌کنیم.
اجرای یک Producer و ارسال پیام به Kafka:
kafka-console-producer.sh ابزاریست که با استفاده از آن می‌توانید به راحتی یک Producer را ایجاد کنید. در terminalی که قبلا برای مدیریت Topicها باز کرده‌اید با استفاده از دستور زیر، Producer را اجرا می‌کنیم:
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic userChat
توضیحات Optionها و آرگومانهایی که در دستور ایجاد Producer استفاده شدند به شرح زیر است:
broker-list localhost:9092--:
نشان دهنده لیست Brokerها میباشد و در صورتیکه تعداد آنها بیش از یک باشد، می‌توان آنها را با "," از هم جدا کرد (پورت پیشفرض Kafka server برای دریافت دستورات، 9092 میباشد).
topic userChat--:
Topicی از Kafka server را مشخص می‌کند که این Producer میخواهد پیامهای خود را برایش ارسال کند.
پس از اجرای دستور فوق، با تایپ کردن و زدن کلید Enter، پیام مورد نظر به Broker موردنظر یعنی localhost:9092 ارسال و در تنها Partition تاپیک userChat ذخیره می‌شود.
اجرای یک Consumer و دریافت پیامهای موجود در Topic مورد نظر:
kafka-console-consumer.sh ابزاریست که خود Kafka در اختیار ما قرار داده است و با استفاده از آن می‌توانیم به راحتی یک Consumer برای userChat ،Topic ایجاد کنیم.
در یک terminal جدید با استفاده از دستور زیر یک Consumer را ایجاد کنید که userChat ،Topic را Subscribe می‌کند:
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic userChat --from-beginning
توضیحات Optionها و آرگومانهایی که در دستور ایجاد یک Consumer استفاده شدند به شرح زیر است:
bootstrap-server localhost:9092-- :
مشخص کننده Kafka serverی است که می‌خواهیم از طریق آن به تمامی اعضای Cluster دسترسی داشته باشیم. در صورتیکه بیش از یک سرور را بخواهیم وارد کنیم، باید آنها را با "," از هم جدا کنیم.
topic userChat --:
مشخص کننده Topicی است که این Consumer میخواهد روی آن Subscribe کند.
from-beginning--:
مشخص کننده این است که Consumer ایجاد شده میخواهد تمامی پیامهای موجود در userChat ، Topic را از اولین offset تا آخرین offset دریافت کند.
پس از اجرای کد فوق مشاهده می‌کنید که پیامهایی که قبلا Producer در تاپیک userChat ثبت کرده‌است، برای این Consumer ارسال می‌شوند.
از اینجا به بعد هر لحظه که در terminal ارسال کننده یا Producer پیامی تایپ کنید و کلید Enter را بزنید، بلافاصله پیام مورد نظر برای Consumer ارسال می‌شود و در terminal آن نمایش داده می‌شود. حتی می‌توانیم چندین Consumer را در terminal های مختلفی اجرا کنیم. در این صورت با ارسال هر پیام از طرف Producer، تمامی Consumerها آن را نمایش می‌دهند.
با استفاده از سادگی راه اندازی و قابلیتهای بسیار زیادی که Apache Kafka در مدیریت جریان داده‌ای بین سیستم‌هایمان به ما می‌دهد، می‌توانیم به سادگی در سیستم‌هایمان قابلیتهای مقیاس پذیری افقی، تحمل در برابر خطا، در دسترس بودن، کارآیی بالا و سادگی مدیریت ارتباطات بخشهای مختلف را اضافه کنیم. در بخشهای بعدی به نحوه ایجاد یک Cluster و اینکه چگونه می‌توان از این بستر در Net. استفاده کرد، می‌پردازیم.
> HTTP/2 403
> Date: Tue, 20 Aug 2013 14:50:41 GMT
> x-ratelimit-limit: 60
> x-ratelimit-remaining: 0
> x-ratelimit-used: 60
> x-ratelimit-reset: 1377013266
> {
> "message": "API rate limit exceeded for xxx.xxx.xxx.xxx. (But here's the good news: Authenticated requests get a higher rate limit. Check out the documentation for more details.)",
> "documentation_url": "https://docs.github.com/rest/overview/resources-in-the-rest-api#rate-limiting"
> } 
namespace ControllerExtensibility.Models
{
public class Result
{
public string ControllerName { get; set; }
public string ActionName { get; set; }
}
} @model ControllerExtensibility.Models.Result
@{
Layout = null;
}
<!DOCTYPE html>
<html>
<head>
<meta name="viewport" content="width=device-width" />
<title>Result</title>
</head>
<body>
<div>Controller: @Model.ControllerName</div>
<div>Action: @Model.ActionName</div>
</body>
</html> using ControllerExtensibility.Models;
using System.Web.Mvc;
namespace ControllerExtensibility.Controllers
{
public class ProductController : Controller
{
public ViewResult Index()
{
return View("Result", new Result
{
ControllerName = "Product",
ActionName = "Index"
});
}
public ViewResult List()
{
return View("Result", new Result
{
ControllerName = "Product",
ActionName = "List"
});
}
}
} using System.Web.Mvc;
namespace ControllerExtensibility.Controllers
{
public class CustomerController : Controller
{
public ViewResult Index()
{
return View("Result", new Result
{
ControllerName = "Customer",
ActionName = "Index"
});
}
public ViewResult List()
{
return View("Result", new Result
{
ControllerName = "Customer",
ActionName = "List"
});
}
}
} using System.Web.Routing;
using System.Web.SessionState;
namespace System.Web.Mvc
{
public interface IControllerFactory
{
IController CreateController(RequestContext requestContext,
string controllerName);
SessionStateBehavior GetControllerSessionBehavior(RequestContext requestContext,
string controllerName);
void ReleaseController(IController controller);
}
} using System;
using System.Web.Mvc;
using System.Web.Routing;
using System.Web.SessionState;
using ControllerExtensibility.Controllers;
namespace ControllerExtensibility.Infrastructure
{
public class CustomControllerFactory : IControllerFactory
{
public IController CreateController(RequestContext requestContext,
string controllerName)
{
Type targetType = null;
switch (controllerName)
{
case "Product":
targetType = typeof (ProductController);
break;
case "Customer":
targetType = typeof (CustomerController);
break;
default:
requestContext.RouteData.Values["controller"] = "Product";
targetType = typeof (ProductController);
break;
}
return targetType == null
? null
: (IController) DependencyResolver.Current.GetService(targetType);
}
public SessionStateBehavior GetControllerSessionBehavior(RequestContext
requestContext, string controllerName)
{
return SessionStateBehavior.Default;
}
public void ReleaseController(IController controller)
{
IDisposable disposable = controller as IDisposable;
if (disposable != null)
{
disposable.Dispose();
}
}
}
} | نام | نوع | توضیحات |
| HttpContext | HttpContextBase | حاوی اطلاعاتی در خصوص درخواست است. |
| RouteData | RouteData | حاوی اطلاعاتی در خصوص Rout است که با درخواست رسیده همخوانی دارد. |
default: requestContext.RouteData.Values["controller"] = "Product"; targetType = typeof(ProductController); break;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using System.Web.Http;
using System.Web.Mvc;
using System.Web.Routing;
using ControllerExtensibility.Infrastructure;
namespace ControllerExtensibility
{
public class MvcApplication : System.Web.HttpApplication
{
protected void Application_Start()
{
AreaRegistration.RegisterAllAreas();
WebApiConfig.Register(GlobalConfiguration.Configuration);
FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
RouteConfig.RegisterRoutes(RouteTable.Routes);
ControllerBuilder.Current.SetControllerFactory(new
CustomControllerFactory());
}
}
} 