java - 如何在 JGROUPS 中禁用 FIFO 和重传协议(protocol)？

Question

我是 Jgroups 的新手，但根据我对文档的理解，它的主要优点之一是可以禁用不需要/不想要的协议(protocol)元素(以实现更好的性能)。但是，当我尝试禁用与“先进先出”交货顺序和“保证交货”有关的任何内容时，出现以下错误:

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>

如果我注释掉最后一个协议(protocol)(pgcast.GMS 协议(protocol))，我不会收到错误，并且它“似乎”可以在单个 Windows VM(在 Google Cloud 上)上工作，但是如果我启动第二个 jvm(仍在同一台 Windows 计算机上)，那么我注意到每个 jvm 都位于“单独的”集群中并且看不到另一个。 (在“正常的tcp.xml”配置中(包括NACKA和XXXX协议(protocol))，例如

<pbcast.NAKACK2 use_mcast_xmit="false"
                   discard_delivered_msgs="true"
                    stats = "false"/>
    <UNICAST3 stats = "false"/>
    <!--<pbcast.STABLE desired_avg_gossip="50000"-->
                   <!--max_bytes="4M"/>-->

一切都“按预期”工作，即，如果我在同一台 Windows 计算机上启动第二个 JVM，则第二个 JVM 确实会加入第一个 JVM 的集群，因此第二个 JVM 上发送的消息会出现在第一个 JVM 中，反之亦然。

那么，有没有一种方法可以禁用 UNICAST3 和 NAKACK2(本质上是与 FIFO 排序或保证消息传递有关的任何内容)，但仍然包含确保“工作完整集群”所需的逻辑，该逻辑还捕获哪些节点离开/加入集群(例如 pbcast.GMS 逻辑？)我不知道如何......

(背景信息:我正在尝试提高性能，我怀疑性能稍慢是因为“保证消息传递”和“FIFO”协议(protocol)，我认为我不需要这些协议(protocol)，因为a)我正在使用TCP，b)消息可以按任何顺序发送。 (也就是说，我假设 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();
    }
}

Answer 1

我没有找到禁用所有可靠消息传输协议(protocol)的方法，因为至少 GMS 协议(protocol)依赖于 NAKACK2。 GMS 要求上述协议(protocol)提供 GET_DIGEST 事件，该事件由 NAKACK2 提供。

但是删除 UNICAST3 协议(protocol)确实有很大帮助，而且现在性能好多了。