BIO'dan NIO'ya ve Reactor modeline JAVA I / O modeli
BIO'dan NIO'ya ve Reactor modeline Java I / O modeli
BIO'dan NIO'ya ve Reactor moduna
Eşzamanlı ve eşzamansız
Senkron I / O Her istek tek tek işlenmelidir.Tek bir talebin işlenmesi, tüm sürecin geçici olarak beklemesine neden olur ve bu olaylar aynı anda yürütülemez. Kullanıcı iş parçacığı bir G / Ç isteği başlattıktan sonra, devam etmeden önce G / Ç işleminin tamamlanması için çekirdeği beklemesi veya yoklaması gerekir.
Eşzamansız G / Ç Aynı anda birden çok istek yürütülebilir ve bir isteğin veya görevin yürütülmesi tüm sürecin geçici olarak beklemesine neden olmaz. Kullanıcı iş parçacığı G / Ç talebini başlattıktan sonra, yürütme devam eder Çekirdek G / Ç işlemi tamamlandığında, kullanıcı iş parçacığı bilgilendirilecek veya kullanıcı iş parçacığı tarafından kaydedilen geri arama işlevi çağrılacaktır.
blok Belli bir talep gönderildikten sonra, talep edilen işlem için gerekli şartlar yerine getirilmediğinden, talep edilen işlem bloke edilmiştir ve şartlar sağlanana kadar geri dönülmeyecektir.
Engellemeyen Talep gönderildikten sonra, talebin gerektirdiği koşullar yerine getirilmezse, sonsuza kadar beklemek yerine, koşulların yerine getirilmediğini bildirmek için derhal bayrak mesajı geri gönderilecektir. Genel olarak, istek sonucunu elde etmek için talep koşulunun bir döngü aracılığıyla karşılanıp karşılanmadığını belirlemek gerekir.
Engellemenin senkronizasyona eşdeğer olmadığı ve engellememenin eşzamansız ile eşdeğer olmadığı unutulmamalıdır. Aslında, bu iki kavram G / Ç modelinde iki farklı boyutu tanımlar.
Senkronizasyon ve zaman uyumsuzluğu, daha sonra başlatılan görevin, birden çok görevin yürütülmesi sırasında ilk olarak başlatılan görevin tamamlanmasını bekleyip beklememesine odaklanır. İlk olarak başlatılan görev isteğinin engellenmesi ve tamamlanmasını beklemesi veya isteğin başarılı olmasını beklemek için hemen döngü boyunca geri dönmesi fark etmeksizin.
Engellemenin ve engellememenin odak noktası, istenen yöntemin hemen dönüp dönmediği (veya koşul karşılanmadığında engellenip engellenmediğidir).
Unix altında beş G / Ç modeli
Unix altında beş G / Ç modeli vardır:
- G / Ç'yi engelleme
- Engellemesiz G / Ç
- G / Ç çoğullama (seçme ve yoklama)
- Sinyale Dayalı G / Ç (SIGIO)
- Eşzamansız G / Ç (Posix.1 aio_ serisi işlevleri)
G / Ç'yi engelleme
Yukarıda belirtildiği gibi, hemen tamamlanamayan G / Ç isteklerinin engellenmesi engellenmeye devam edecektir. G / Ç'yi bloke etme aşağıdaki iki aşamaya bölünmüştür.
- Aşama 1: Verilerin hazır olmasını bekleyin. Ağ G / Ç durumu, uzak verilerin birbiri ardına gelmesini beklemektir; disk G / Ç durumu, disk verilerinin diskten çekirdek modu belleğine okunmasını beklemektir.
- Aşama 2: Veri kopyalama. Sistem güvenliği için, kullanıcı modu programlarının çekirdek modu belleğini doğrudan okuma izni yoktur, bu nedenle çekirdek modu belleğindeki verilerin kullanıcı modu belleğine kopyalanmasından çekirdek sorumludur.
Engellemesiz G / Ç
Engellemeyen G / Ç istekleri aşağıdaki üç aşamayı içerir
- Yuva, çekirdeğe istenen G / Ç işlemi tamamlanamadığında iş parçacığını uyutmamasını, ancak isteğin engellenmemesi için bir hata kodu (EWOULDBLOCK) döndürmesini söylemek için NONBLOCK (engellemesiz) olarak ayarlanmıştır.
- G / Ç çalıştırma işlevi, verilerin hazır olup olmadığını sürekli olarak test edecek, yoksa veriler hazır olana kadar teste devam edecektir. Tüm G / Ç talep sürecinde, kullanıcı iş parçacığı her bir G / Ç isteği başlatıldıktan hemen sonra geri dönebilse de, verileri beklemek için, yine de çok fazla CPU kaynağı tüketen talebi sürekli olarak sorgulaması ve tekrarlaması gerekir.
- Veriler hazırdır ve çekirdekten kullanıcı alanına kopyalanır.
Genel olarak, bu model doğrudan nadiren kullanılır, ancak diğer G / Ç modellerinde engellemesiz G / Ç kullanılır. Bu yöntemin tek bir G / Ç isteği için çok az anlamı vardır, ancak G / Ç çoğullaması için koşullar sağlar.
G / Ç çoğullama
G / Ç çoğullama seçme veya sorgulama işlevlerini kullanacaktır, bu iki işlev de iş parçacığı bloke edecektir, ancak G / Ç'yi engellemenin aksine, bu iki işlev aynı anda birden çok G / Ç işlemini engelleyebilir. Ve aynı anda birden çok okuma işleminin ve birden çok yazma işleminin G / Ç işlevlerini algılayabilir ve okunabilir veya yazılabilir veriler olana kadar yalnızca G / Ç işlem işlevini çağırabilir.
Süreç bakış açısından, I / O talepleri için seçme işlevinin kullanılması ile eşzamanlı engelleme modeli arasında çok fazla fark yoktur.İzleme kanallarının eklenmesi ve ekstra iş ekleyen seçme işlevinin çağrılması gibi ek işlemler bile vardır. Ancak, select kullanmanın en büyük avantajı, kullanıcıların tek bir iş parçacığında aynı anda birden çok Kanal G / Ç isteğini işleyebilmesidir. Kullanıcı birden fazla Kanalı kaydedebilir ve ardından etkinleştirilen Kanalları okumak için sürekli olarak seçim çağrısı yapabilir, böylece birden fazla I / O talebi aynı iş parçacığında aynı anda işlenebilir. Eşzamanlı engelleme modelinde, bu hedefe çok iş parçacıklı okuma yoluyla ulaşılmalıdır.
Select / anket çağrılırken, belirli bir aşama 1'in verileri hazır olana kadar birden fazla Kanalı sorgulamaktan bir kullanıcı modu dizisi sorumludur ve ardından gerçek kullanıcı dizisi, aşama 2'nin kopyasını yürütmek için bildirilir. Engellemesiz G / Ç sorgulaması gerçekleştirmek için özel bir kullanıcı modu iş parçacığı aracılığıyla simülasyon, birinci fazın asenkronizasyonunu gerçekleştirir.
Sinyale Dayalı G / Ç (SIGIO)
İlk olarak, soketin sinyal güdümlü G / Ç gerçekleştirmesine ve bir sinyal işleme fonksiyonu kurmasına izin veriyoruz, iş parçacığı bloke olmadan çalışmaya devam ediyor. Veriler hazır olduğunda, iş parçacığı bir SIGIO sinyali alır ve verileri işlemek için sinyal işleme işlevinde G / Ç işlem işlevini çağırabilir.
Eşzamansız G / Ç
Çekirdeğe açıklama sözcüğünü, arabellek işaretçisini, arabellek boyutunu, dosya uzaklığını ve bildirim yöntemini söylemek için aio_read işlevini çağırın ve ardından hemen geri dönün. Çekirdek, verileri arabelleğe kopyaladıktan sonra, uygulamaya bildirir. Bu nedenle, eşzamansız G / Ç modunda, aşama 1 ve aşama 2'nin tümü çekirdek tarafından tamamlanır ve tamamlanma, kullanıcı iş parçacıklarının katılımını gerektirmez.
Asenkron G / Ç haricinde, diğer dört modelin 2. aşaması temelde aynıdır ve hepsi çekirdek modundan kullanıcı moduna veri kopyalar. Aradaki fark, 1. aşamanın farklı olmasıdır. İlk dördü senkron I / O'dur.
Java'da dört G / Ç modeli
Önceki bölümde bahsedilen Unix'te beş G / Ç modeli vardır.Java, sinyal güdümlü G / Ç'ye ek olarak diğer dört G / Ç modelini de destekler. Bunlar arasında, Java tarafından sağlanan en eski engelleme G / Ç'si G / Ç'yi engellemektir ve NIO, engellemeyen G / Ç'dir.Aynı zamanda, NIO tarafından uygulanan Reaktör modu, G / Ç çoğullama modelinin gerçekleştirilmesidir ve AIO tarafından uygulanan Proactor modu Asenkron I / O modelinin gerçekleştirilmesidir.
IO'dan NIO'ya
Akış odaklı ve tampon odaklı
Java IO akış odaklıdır. Tüm baytlar okunana kadar akıştan (InputStream / OutputStream) bir veya daha fazla bayt okunduğunda, bunlar herhangi bir yerde önbelleğe alınmaz. Ayrıca, akıştaki verileri ileri ve geri hareket ettiremez.İleri ve geri hareket etmesi gerekiyorsa, önce bir arabellekte önbelleğe alınması gerekir.
Java NIO arabellek odaklıdır ve veriler bir arabelleğe okunacaktır ve gerektiğinde arabellekte ileri geri hareket ettirilebilir, bu da işleme sürecinin esnekliğini artırır. Ancak aynı zamanda, arabelleği işlemeden önce, arabelleğin işlenmesi gereken verileri içerip içermediğini kontrol etmek ve arabelleğe daha fazla veri okunduğunda, arabellekteki işlenmemiş verilerin üzerine yazılmayacağından emin olmak gerekir.
Engelleyen ve engellemeyen
Çeşitli Java IO akışları engellenir. Bir iş parçacığı read () veya write () yöntemini çağırdığında, veri okunana veya veriler tamamen yazılana kadar iş parçacığı engellenir. İş parçacığı, engelleme süresi boyunca başka hiçbir şeyi işleyemez.
Java NIO, engellemeyen bir moddur. Okuma ve yazma istekleri mevcut iş parçacığını engellemez Mevcut iş parçacığı, veriler okunmadan / yazılmadan önce başka şeyler yapmaya devam edebilir, böylece tek bir iş parçacığı birden fazla giriş ve çıkış kanalını yönetebilir.
Seçici
Java NIO seçicisi, tek bir iş parçacığının aynı anda birden fazla kanalı izlemesine izin verir.Aynı seçiciye birden fazla kanal kaydedebilir ve ardından hazır olan kanalları "seçmek" için tek bir iş parçacığı kullanabilirsiniz. Bu "seçim" mekanizması, tek bir iş parçacığının birden fazla kanalı yönetmesini mümkün kılar.
Sıfır kopya
Java NIO'da sağlanan FileChannel, FileChannel'daki verileri doğrudan başka bir Kanala kopyalayabilen veya başka bir Kanaldaki verileri doğrudan Dosya Kanalına kopyalayabilen iki yöntem transferTo ve transferFrom'a sahiptir.
Bu arabirim genellikle verimli ağ / disk denetleyicisi veri iletimi ve büyük dosya kopyalama için kullanılır. İşletim sistemi desteği olması durumunda, bu yöntemle veri aktarımı, kaynak verilerin çekirdek modundan kullanıcı moduna ve ardından kullanıcı modundan hedef kanalın çekirdek moduna kopyalanmasına gerek olmadığı gibi iki kullanıcı modu ve çekirdek modundan da kaçınıyor. Arasında bağlam geçişi, yani "sıfır kopya" kullanılır, bu nedenle performansı genellikle Java IO'da sağlanan yöntemden daha yüksektir.
FileChannel'in sıfır kopyasını kullanarak yerel dosya içeriğini ağa aktarmak için örnek kod aşağıda gösterilmiştir.
public class NIOClient {
public static void main (String args), IOException, InterruptedException {atar
SocketChannel socketChannel = SocketChannel.open ();
InetSocketAddress adresi = yeni InetSocketAddress (1234);
socketChannel.connect (adres);
RandomAccessFile dosyası = yeni RandomAccessFile (
NIOClient.class.getClassLoader (). GetResource ("test.txt"). GetFile (), "rw");
FileChannel channel = file.getChannel ();
channel.transferTo (0, channel.size (), socketChannel);
channel.close ();
file.close ();
socketChannel.close ();
}
}
Tek iş parçacığı engelleyen G / Ç sunucusu
Engelleme G / Ç kullanan bir sunucu, genellikle bağlantı isteklerini kabul etmek ve verileri tek tek okumak ve ardından sonraki isteği işlemek için bir döngü kullanır.
public class IOServer {
private static final Logger LOGGER = LoggerFactory.getLogger (IOServer.class);
public static void main (String args) {
ServerSocket serverSocket = boş;
Deneyin {
serverSocket = new ServerSocket ();
serverSocket.bind (yeni InetSocketAddress (2345));
} catch (IOException ex) {
LOGGER.error ("Dinlenemedi", örn.);
dönüş;
}
Deneyin{
while (true) {
Soket soketi = serverSocket.accept ();
InputStream inputstream = socket.getInputStream ();
LOGGER.info ("Alınan ileti {}", IOUtils.toString (yeni InputStreamReader (inputstream)));
}
} catch (IOException ex) {
Deneyin {
serverSocket.close ();
} catch (IOException e) {
}
LOGGER.error ("Mesaj okunamadı", örn.);
}
}
}
Her istek için bir ileti dizisi oluşturun
Yukarıdaki örnek, tüm istekleri tek tek işlemek için tek bir iş parçacığı kullanır.Aynı anda yalnızca bir istek işlenebilir.G / Ç için bekleme süreci çok fazla CPU kaynağını boşa harcar ve birden fazla CPU'nun avantajlarını tam olarak kullanamaz. Aşağıda, çoklu iş parçacığı kullanan engelleme G / Ç modelinde bir iyileştirme verilmiştir. Bir bağlantı başarıyla kurulduktan sonra, G / Ç işlemlerini yürütmek için ayrı bir iş parçacığı oluşturulur.
public class IOServerMultiThread {
private static final Logger LOGGER = LoggerFactory.getLogger (IOServerMultiThread.class);
public static void main (String args) {
ServerSocket serverSocket = boş;
Deneyin {
serverSocket = new ServerSocket ();
serverSocket.bind (yeni InetSocketAddress (2345));
} catch (IOException ex) {
LOGGER.error ("Dinlenemedi", örn.);
dönüş;
}
Deneyin{
while (true) {
Soket soketi = serverSocket.accept ();
yeni Konu( ()- > {
Deneyin{
InputStream inputstream = socket.getInputStream ();
LOGGER.info ("Alınan ileti {}", IOUtils.toString (yeni InputStreamReader (inputstream)));
} catch (IOException ex) {
LOGGER.error ("Mesaj okunamadı", örn.);
}
}).Başlat();
}
} catch (IOException ex) {
Deneyin {
serverSocket.close ();
} catch (IOException e) {
}
LOGGER.error ("Bağlantı başarısız oldu", örn.);
}
}
}
İstekleri işlemek için iş parçacığı havuzunu kullanın
Çok fazla bağlantı isteğini önlemek için, sunucu çok fazla iş parçacığı oluşturur ve aşırı iş parçacığı bağlamı değiştirmenin ek yüküne neden olur. Oluşturulan iş parçacığı sayısı, aşağıda gösterildiği gibi iş parçacığı havuzu ile sınırlandırılabilir.
public class IOServerThreadPool {
private static final Logger LOGGER = LoggerFactory.getLogger (IOServerThreadPool.class);
public static void main (String args) {
ExecutorService executorService = Executors.newFixedThreadPool (Runtime.getRuntime (). AvailableProcessors ());
ServerSocket serverSocket = boş;
Deneyin {
serverSocket = new ServerSocket ();
serverSocket.bind (yeni InetSocketAddress (2345));
} catch (IOException ex) {
LOGGER.error ("Dinlenemedi", örn.);
dönüş;
}
Deneyin{
while (true) {
Soket soketi = serverSocket.accept ();
executorService.submit (() - > {
Deneyin{
InputStream inputstream = socket.getInputStream ();
LOGGER.info ("Alınan ileti {}", IOUtils.toString (yeni InputStreamReader (inputstream)));
} catch (IOException ex) {
LOGGER.error ("Mesaj okunamadı", örn.);
}
});
}
} catch (IOException ex) {
Deneyin {
serverSocket.close ();
} catch (IOException e) {
}
LOGGER.error ("Bağlantı başarısız oldu", örn.);
}
}
}
Klasik Reaktör modu
Klasik Reaktör modu diyagramı aşağıda gösterilmiştir.
Reactor modunda, aşağıdaki roller dahildir
- Reaktör I / O olaylarını ilgili İşleyiciye gönder
- Akseptör İstemci bağlantı isteklerini işleme
- İşleyiciler Engellemesiz okuma / yazma gerçekleştirin
Reactor modunun en basit uygulama kodu aşağıda gösterilmiştir.
public class NIOServer {
private static final Logger LOGGER = LoggerFactory.getLogger (NIOServer.class);
public static void main (String argümanları) IOException {atar
Seçici seçici = Selector.open ();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open ();
serverSocketChannel.configureBlocking (false);
serverSocketChannel.bind (yeni InetSocketAddress (1234));
serverSocketChannel.register (seçici, SelectionKey.OP_ACCEPT);
while (selector.select () > 0) {
Ayarlamak < Seçim Anahtarı > anahtarlar = selector.selectedKeys ();
Yineleyici < Seçim Anahtarı > iterator = keys.iterator ();
while (iterator.hasNext ()) {
SelectionKey tuşu = iterator.next ();
iterator.remove ();
if (key.isAcceptable ()) {
ServerSocketChannel acceptServerSocketChannel = (ServerSocketChannel) key.channel ();
SocketChannel socketChannel = acceptServerSocketChannel.accept ();
socketChannel.configureBlocking (yanlış);
LOGGER.info ("{} 'den gelen isteği kabul et", socketChannel.getRemoteAddress ());
socketChannel.register (seçici, SelectionKey.OP_READ);
} else if (key.isReadable ()) {
SocketChannel socketChannel = (SocketChannel) key.channel ();
ByteBuffer tamponu = ByteBuffer.allocate (1024);
int count = socketChannel.read (arabellek);
eğer (say < = 0) {
socketChannel.close ();
key.cancel ();
LOGGER.info ("Geçersiz veri alındı, bağlantıyı kapatın");
devam et;
}
LOGGER.info ("Alınan mesaj {}", yeni Dize (buffer.array ()));
}
keys.remove (anahtar);
}
}
}
}
Okumayı kolaylaştırmak için yukarıda gösterilen kod, Reactor modelindeki tüm rolleri tek bir sınıfa koyar.
Yukarıda gösterilen koddan da görebileceğiniz gibi, aynı Selector nesnesine birden çok Kanal kaydedilebilir, bu da bir iş parçacığının aynı anda birden fazla istek durumunu (Kanal) izlemesini sağlar. Aynı zamanda, kayıt olurken, ilgilendiği olayı belirtmeniz gerekir.Örneğin, yukarıdaki koddaki socketServerChannel nesnesi yalnızca OP_ACCEPT olayını kaydeder ve socketChannel nesnesi yalnızca OP_READ olayını kaydeder.
selector.select () engelliyor. Bu yöntem, en az bir kanal mevcut olduğunda kullanılabilir kanalların sayısını döndürür. Aynı zamanda, bu yöntem yalnızca Kanal kaydı sırasında belirtilen ilgili olayları yakalar.
Çok işçili Reaktör modu
Klasik Reactor modunda, bir iş parçacığı aynı anda birden fazla isteği (Kanal) izleyebilse de, tüm okuma / yazma istekleri ve yeni bağlantı isteklerinin işlenmesi aynı iş parçacığı içinde işlenir.Birden fazla CPU'dan tam olarak yararlanmak ve aynı anda okumak istemiyorum. / Write işlemi ayrıca yeni bağlantı isteklerinin işlenmesini de engeller. Bu nedenle, çoklu okuma / yazma işlemlerini aşağıdaki şekilde gösterildiği gibi paralel olarak işlemek için çoklu okuma tanıtılabilir.
Çok parçacıklı Reaktör modunun örnek kodu aşağıda gösterilmiştir.
public class NIOServer {
private static final Logger LOGGER = LoggerFactory.getLogger (NIOServer.class);
public static void main (String argümanları) IOException {atar
Seçici seçici = Selector.open ();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open ();
serverSocketChannel.configureBlocking (false);
serverSocketChannel.bind (yeni InetSocketAddress (1234));
serverSocketChannel.register (seçici, SelectionKey.OP_ACCEPT);
while (true) {
eğer (selector.selectNow () < 0) {
devam et;
}
Ayarlamak < Seçim Anahtarı > anahtarlar = selector.selectedKeys ();
Yineleyici < Seçim Anahtarı > iterator = keys.iterator ();
while (iterator.hasNext ()) {
SelectionKey tuşu = iterator.next ();
iterator.remove ();
if (key.isAcceptable ()) {
ServerSocketChannel acceptServerSocketChannel = (ServerSocketChannel) key.channel ();
SocketChannel socketChannel = acceptServerSocketChannel.accept ();
socketChannel.configureBlocking (yanlış);
LOGGER.info ("{} 'den gelen isteği kabul et", socketChannel.getRemoteAddress ());
SelectionKey readKey = socketChannel.register (seçici, SelectionKey.OP_READ);
readKey.attach (yeni İşlemci ());
} else if (key.isReadable ()) {
İşlemci işlemci = (İşlemci) key.attachment ();
işlemci.process (anahtar);
}
}
}
}
}
Yukarıda gösterilen koddan da görebileceğiniz gibi, SocketChannel'in OP_READ olayını kaydettikten sonra, karşılık gelen SelectionKey'e bir nesne ekleyebilirsiniz (bu örnekte, okuma talebini işleyen bir İşlemci nesnesi eklenmiştir) ve okunabilir olayı elde ettikten sonra , Nesneyi kaldırabilirsiniz.
Not: Bir nesnenin takılması ve nesnenin kaldırılması NIO tarafından sağlanan bir işlemdir, ancak bu işlem Reactor modunda gerekli bir işlem değildir.Bu makale, onu yalnızca NIO arayüzünün gösterimini kolaylaştırmak için kullanır.
Spesifik okuma isteği işleme, aşağıda gösterildiği gibi İşlemci sınıfındadır. Tüm istekleri işlemek için bu sınıfta bir statik iş parçacığı havuzu oluşturulur. İşlem yöntemi, G / Ç isteklerini doğrudan işlemez, ancak G / Ç işlemini işleme için yukarıdaki iş parçacığı havuzuna gönderir; bu, çoklu iş parçacığı avantajlarından tam olarak yararlanır ve yeni bağlantıları ve okuma / yazma işlemlerini işleyecektir. İşleme, farklı iş parçacıkları içine yerleştirilir ve okuma / yazma işlemi artık yeni bağlantı isteklerinin işlenmesini engellemez.
public class İşlemci {
private static final Logger LOGGER = LoggerFactory.getLogger (Processor.class);
private static final ExecutorService service = Executors.newFixedThreadPool (16);
public void process (SelectionKey selectionKey) {
service.submit (() - > {
ByteBuffer tamponu = ByteBuffer.allocate (1024);
SocketChannel socketChannel = (SocketChannel) selectionKey.channel ();
int count = socketChannel.read (arabellek);
eğer (say < 0) {
socketChannel.close ();
selectionKey.cancel ();
LOGGER.info ("{} \ t Okuma bitti", socketChannel);
boş döndür;
} else if (count == 0) {
boş döndür;
}
LOGGER.info ("{} \ t Mesaj oku {}", socketChannel, new String (buffer.array ()));
boş döndür;
});
}
}
Çoklu Reaktör
Netty'de kullanılan Reaktör modu, birden fazla Reaktör sunar; yani, tüm bağlantı taleplerini izlemekten bir ana Reaktör sorumludur ve okuma / yazma taleplerini izlemek ve işlemekten çok sayıda çocuk Reaktör sorumludur, bu da ana Reaktör üzerindeki basıncı ve ana Reaktör üzerindeki basıncı azaltır. Ve gecikmeye neden oldu.
Ve her çocuk Reactor bağımsız bir iş parçacığına aittir ve başarıyla bağlanan her Kanalın tüm işlemleri aynı iş parçacığı tarafından işlenir. Bu, aynı isteğin tüm durumlarının ve bağlamlarının aynı iş parçacığında olmasını sağlar, gereksiz bağlam değiştirmeyi önler ve ayrıca istek yanıt durumunun izlenmesini kolaylaştırır.
Çoklu reaktör modunun şematik diyagramı aşağıda gösterilmiştir.
Multi-Reactor örnek kodu aşağıda gösterilmiştir.
public class NIOServer {
private static final Logger LOGGER = LoggerFactory.getLogger (NIOServer.class);
public static void main (String argümanları) IOException {atar
Seçici seçici = Selector.open ();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open ();
serverSocketChannel.configureBlocking (false);
serverSocketChannel.bind (yeni InetSocketAddress (1234));
serverSocketChannel.register (seçici, SelectionKey.OP_ACCEPT);
int coreNum = Runtime.getRuntime (). availableProcessors ();
İşlemci işlemcileri = yeni İşlemci;
for (int i = 0; i < processors.length; i ++) {
işlemciler = yeni İşlemci ();
}
int indeksi = 0;
while (selector.select () > 0) {
Ayarlamak < Seçim Anahtarı > anahtarlar = selector.selectedKeys ();
for (SelectionKey key: keys) {
keys.remove (anahtar);
if (key.isAcceptable ()) {
ServerSocketChannel acceptServerSocketChannel = (ServerSocketChannel) key.channel ();
SocketChannel socketChannel = acceptServerSocketChannel.accept ();
socketChannel.configureBlocking (yanlış);
LOGGER.info ("{} 'den gelen isteği kabul et", socketChannel.getRemoteAddress ());
İşlemci işlemci = işlemciler;
işlemci.addChannel (socketChannel);
}
}
}
}
}
Yukarıdaki kodda gösterildiği gibi, bu makalede belirlenen alt reaktör sayısı, mevcut makinede bulunan çekirdek sayısının iki katıdır (Netty'deki varsayılan alt reaktör sayısı ile tutarlıdır). Başarıyla bağlanan her SocketChannel için, döngüsel olarak farklı alt Reaktörlere devredilir.
SocketChannel'in alt Reaktör tarafından işlenmesi aşağıdaki gibidir.
public class İşlemci {
private static final Logger LOGGER = LoggerFactory.getLogger (Processor.class);
private static final ExecutorService hizmeti =
Executors.newFixedThreadPool (2 * Runtime.getRuntime (). AvailableProcessors ());
özel Seçici seçici;
public Processor (), IOException {
this.selector = SelectorProvider.provider (). openSelector ();
Başlat();
}
public void addChannel (SocketChannel socketChannel) ClosedChannelException {
socketChannel.register (this.selector, SelectionKey.OP_READ);
}
public void başlangıcı () {
service.submit (() - > {
while (true) {
eğer (selector.selectNow () < = 0) {
devam et;
}
Ayarlamak < Seçim Anahtarı > anahtarlar = selector.selectedKeys ();
Yineleyici < Seçim Anahtarı > iterator = keys.iterator ();
while (iterator.hasNext ()) {
SelectionKey tuşu = iterator.next ();
iterator.remove ();
eğer (key.isReadable ()) {
ByteBuffer tamponu = ByteBuffer.allocate (1024);
SocketChannel socketChannel = (SocketChannel) key.channel ();
int count = socketChannel.read (arabellek);
eğer (say < 0) {
socketChannel.close ();
key.cancel ();
LOGGER.info ("{} \ t Okuma bitti", socketChannel);
devam et;
} else if (count == 0) {
LOGGER.info ("{} \ t Mesaj boyutu 0", socketChannel);
devam et;
} Başka {
LOGGER.info ("{} \ t Mesaj oku {}", socketChannel, new String (buffer.array ()));
}
}
}
}
});
}
}
İşlemci'de, statik bir iş parçacığı havuzu da oluşturulur ve iş parçacığı havuzunun boyutu makine çekirdeği sayısının iki katıdır. Her Processor örneği bir Selector örneği içerir. Aynı zamanda, bir İşlemci vakası her alındığında, iş parçacığı havuzuna bir görev gönderilir ve görev, normal koşullar altında durdurulmadan bir döngü içinde işlenir. İşlemciye gönderilen SocketChannel, Seçicisindeki olayları kaydederek ilgili göreve eklenir. Sonuç olarak, her alt Reaktör bir Selector nesnesi içerir ve bağımsız bir iş parçacığı tarafından işlenir.
