Netty 和 Bukkit 的线程安全

Question

我正在为何时读取通道以及何时在 Netty 中写入通道创建侦听器。监听器是在 Bukkit 的事件系统中处理的，这不是问题，问题是线程安全。 Bukkit Api 在主线程上运行，必须同步，否则基本上服务器会爆炸。 Netty 在多个线程上运行，因此使用 Bukkit api 进行交叉通信可能会很困难。我在 Bukkit 论坛上问过，我得到的最佳答案是创建一个 AtomicBoolean 并有一个 while 循环，这将解决同步取消发送和写入数据包的问题，​​但它不能解决更改数据包的问题正在发送/写入。当在 Bukkit 中调用事件时，会从 @EventHandler 下的每个类调用侦听器。所有这些代码都必须与主线程同步，我不确定在这种情况下我将如何使用同步。抱歉代码格式不好，我不能很好地工作这个系统。

public class ConnectionInjector extends ChannelDuplexHandler {

    private User user;
    private Channel channel;
    private PacketRecieveEvent recieve;
    private PacketSendEvent send;
    private boolean isInjected = false;
    private boolean isClosed = false;

    public ConnectionInjector(User user) {
        this.init(user.getPlayer());
    }
    public void close() {
        if (!this.isClosed) {
            this.isClosed = true;
            if (this.isInjected) {
                getChannel().eventLoop().submit(new Callable<Object>() {
                    @Override
                    public Object call() throws Exception {

                        getChannel().pipeline().remove(ConnectionInjector.this);
                        return null;

                    }

                });

                this.isInjected = false;
            }
        }
    }

    public boolean isInjected() {
        return this.isInjected;
    }

    public boolean isClosed() {
        return this.isClosed;
    }


    @Override
    public void write(ChannelHandlerContext context, Object packet, ChannelPromise channel) {
        if (this.isClosed()){
            throw new IllegalStateException("Connection closed already");
        }
        PacketData d = new PacketData(packet);
        send = new PacketSendEvent(user, d);//the event
        Bukkit.getScheduler().scheduleSyncDelayedTask(Main.getInstance(), new Runnable() {
            public void run() {
                Bukkit.getPluginManager().callEvent(send); //runnable puts it on the main thread
            }
        });//end of bukkit thread back on netty
        if (!send.isCancelled()){// this possible happens before the above code which is very bad
            try {
                if (send.getPacket().getRawPacket() == null){
                    throw new IllegalStateException("sent packet was null: " + send.getPacket().getRawPacket().getClass().getName());
                }
                super.write(context, send.getPacket().getRawPacket(), channel); //possible happens before send is even created.
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext context, final Object packet) throws Exception {
        PacketData p = new PacketData(packet);
        recieve = new PacketRecieveEvent(user, p);
        Bukkit.getScheduler().scheduleSyncDelayedTask(Main.getInstance(), new Runnable() {
            public void run() {
                Bukkit.getPluginManager().callEvent(recieve); 
            }

        });
        if (recieve.isCancelled()){
            super.channelRead(context, recieve.getPacket().getRawPacket());
        }


    }

    public void injectfakePacket(Object packet) {
        if (this.isClosed()){
            throw new IllegalStateException("Injector is closed");
        }
        this.getChannel().pipeline().context("encoder").fireChannelRead(packet);
    }

    public void init(Player player) {
      this.channel = (Channel) Nms.getNetChannel(player);
        this.channel.pipeline().addBefore("packet_handler", "epickitpvp", this);
        this.isInjected = true;
    }

    public Channel getChannel() {
        if (this.channel == null){
            throw new IllegalStateException("channel is null");
        }
        return this.channel;
    }
} 

Answer 1

您可以通过异步线程访问 Bukkit 的调度程序，因此您可以在异步线程内部调度一个新的同步任务，因为同步线程只有在主线程准备好时才会被调用。

复杂的短：用

安排一个新的同步任务

Bukkit.getScheudler().scheudleSyncTask(PLUGIN, new Runnable() {
    public void run() {
        // Bukkit methods here
    }
});

在你的异步线程里面，你很好。

Answer 2

我不完全理解你想用这段代码完成什么，所以我不能给你一个直接的答案，但我可以解决你在线程方面可能遇到的一些问题。

首先，Java 可以并且将在每个线程的基础上缓存变量，这意味着您最终会得到相同变量的多个副本，这些副本具有不同的值，具体取决于您所在的线程。这种缓存可能会导致对于不了解它的人来说有很多困惑。但是，您可以使用 volatile 关键字告诉 Java 使变量在线程之间保持一致状态。更多关于 volatile 的信息可以在这里找到：http://www.javamex.com/tutorials/synchronization_volatile.shtml

如果您在理解 Java 中的 synchronized 关键字时遇到困难，我建议您先对信号量是什么进行一些研究，然后再重新审视同步。很好地理解信号量是什么以及它是如何工作的，将使您更容易理解如何在线程需要共享状态的多线程环境中工作（但是，如果可能的话，您应该避免这种情况）。

您还可能会发现 Java 的并发集合 api 非常有用：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/package-summary.html