Я новичок в Jgroups, но, основываясь на моем понимании документации, одно из его ключевых преимуществ заключается в том, что можно отключить элементы протокола, которые не нужны / не нужны (для достижения лучшей производительности). Однако, когда я попытался отключить все, что связано с порядком доставки "FIFO" и "гарантией доставки", я получил следующую ошибку:
Exception in thread "main" java.lang.Exception: events [GET_DIGEST SET_DIGEST ] are required by GMS, but not provided by any of the protocols below it
at org.jgroups.stack.Configurator.sanityCheck(Configurator.java:320)
at org.jgroups.stack.Configurator.connectProtocols(Configurator.java:197)
at org.jgroups.stack.Configurator.setupProtocolStack(Configurator.java:115)
at org.jgroups.stack.Configurator.setupProtocolStack(Configurator.java:49)
at org.jgroups.stack.ProtocolStack.setup(ProtocolStack.java:475)
at org.jgroups.JChannel.init(JChannel.java:965)
at org.jgroups.JChannel.<init>(JChannel.java:148)
at org.jgroups.JChannel.<init>(JChannel.java:130)
at RpcDispatcherTest.start(RpcDispatcherTest.java:29)
at RpcDispatcherTest.main(RpcDispatcherTest.java:83)
Мой xml-файл конфигурации выглядит так:
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="urn:org:jgroups"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/jgroups.xsd">
<TCP bind_addr="127.0.0.1"
bind_port="7800"
recv_buf_size="${tcp.recv_buf_size:130k}"
send_buf_size="${tcp.send_buf_size:130k}"
max_bundle_size="64K"
sock_conn_timeout="300"
enable_diagnostics="true"
thread_pool.min_threads="10"
thread_pool.max_threads="20"
thread_pool.keep_alive_time="30000"
stats = "false"
/>
<TCPPING initial_hosts="127.0.0.1[7800]"
port_range="0" stats = "false"/>
<MERGE3 min_interval="10000"
max_interval="30000" stats = "false"/>
<FD_SOCK stats = "false"/>
<FD timeout="3000" max_tries="3" stats = "false" />
<VERIFY_SUSPECT timeout="1500" stats = "false" />
<pbcast.GMS print_local_addr="true" join_timeout="2000"
view_bundling="true" stats = "false"/>
</config>
Если я закомментирую последний протокол (pgcast.GMS), я НЕ получаю ошибок, и он "кажется" работает на одной виртуальной машине с Windows (в Google Cloud), но если я запускаю 2-ю jvm ( все еще на той же машине Windows), то я замечаю, что каждый jvm находится в «отдельном» кластере и не видит другого. (В конфигурации «normal tcp.xml» (которая включает протоколы NACKA и XXXX), например,
<pbcast.NAKACK2 use_mcast_xmit="false"
discard_delivered_msgs="true"
stats = "false"/>
<UNICAST3 stats = "false"/>
<!--<pbcast.STABLE desired_avg_gossip="50000"-->
<!--max_bytes="4M"/>-->
все работает "как положено", т. Е. Если я запускаю 2-ю JVM на той же машине с Windows, 2-я JVM действительно присоединяется к кластеру 1-й JVM, и поэтому сообщения, отправленные на 2-й JVM, появляются в 1-й JVM и наоборот. наоборот.
Итак, есть ли способ отключить UNICAST3 и NAKACK2 (по сути, все, что связано с упорядочением FIFO или гарантией доставки сообщений), но все же включает логику, необходимую для обеспечения «работающего полного кластера», который также фиксирует, какие узлы покидают / присоединиться к кластеру (например, логика pbcast.GMS?) Я не мог понять, как ....
(Справочная информация: я пытаюсь улучшить производительность, и подозреваю, что несколько медленная производительность вызвана протоколами "гарантированной доставки сообщений" и "FIFO", которые мне не нужны, потому что а) я используя TCP и б) сообщения можно отправлять в любом порядке. (Тем не менее, я предполагаю, что TCP почти по определению обеспечивает доставку сообщений, поскольку это очень важно.) Я также использую Google Cloud, где, как мне кажется, «гарантирующий» аспект логики TCP работает на высокооптимизированных маршрутизаторы и многоадресная рассылка в любом случае не разрешены, что подавляет одно из основных преимуществ многоадресной рассылки UDP.)
Наконец (и я НЕ думаю, что это необходимо), но вот мой тестовый код (который является лишь небольшой модификацией демонстрации, которая поставляется с JGroups 4.0):
import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.blocks.*;
import org.jgroups.util.RspList;
import org.jgroups.util.Util;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
public class RpcDispatcherTest {
JChannel channel;
RpcDispatcher disp;
RspList rsp_list;
String props = "gs-tcp.xml"; // set by application
public static int print(int number) throws Exception {
return number;
}
public void start() throws Exception {
RequestOptions opts=new RequestOptions(ResponseMode.GET_FIRST, 1000);
channel=new JChannel(props);
disp=new RpcDispatcher(channel, this);
channel.connect("RpcDispatcherTestGroup");
final Address myCurAddress = channel.getAddress();
System.out.println("Currrent address is " + myCurAddress + " all members address are " + channel.getView().getMembers().toString());
final long t1 = System.currentTimeMillis();
final IntStream x = IntStream.range(0, 1_000_000);
final AtomicInteger cnt = new AtomicInteger();
x.asLongStream().parallel().forEach(l -> {
try {
final int i = (int) l;
if (i % (100) == 0) {
System.out.println("At " + i + " on thread + " + Thread.currentThread().getId());
}
final MethodCall call=new MethodCall(getClass().getMethod("print", int.class));
call.setArgs(i);
final CompletableFuture<Integer> response = disp.<Integer>callRemoteMethodWithFuture(myCurAddress, call, opts);
response.thenAccept(integer -> {
if (integer % (1024*8) == 0) {
System.out.println("At " + cnt.incrementAndGet() + " Execution time for " + integer + " is " + (System.currentTimeMillis() - t1)/1000f);
}
});
} catch (Exception e) {
e.printStackTrace();
}
});
// Util.close(disp, channel);
}
public static void main(String[] args) throws Exception {
new RpcDispatcherTest().start();
}
}