百度360必应搜狗淘宝本站头条
当前位置:网站首页 > IT技术 > 正文

Go 中的 channel 与 Java BlockingQueue 的本质区别

wptr33 2025-09-19 03:55 1 浏览

前言

最近在实现两个需求,由于两者之间并没有依赖关系,所以想利用队列进行解耦;但在 Go 的标准库中并没有现成可用并且并发安全的数据结构;但 Go 提供了一个更加优雅的解决方案,那就是 channel

channel 应用

GoJava 的一个很大的区别就是并发模型不同,Go 采用的是 CSP(Communicating sequential processes) 模型;用 Go 官方的说法:

Do not communicate by sharing memory; instead, share memory by communicating.

翻译过来就是:不用使用共享内存来通信,而是用通信来共享内存。

而这里所提到的通信,在 Go 里就是指代的 channel

只讲概念并不能快速的理解与应用,所以接下来会结合几个实际案例更方便理解。

futrue task

Go 官方没有提供类似于 JavaFutureTask 支持:

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Task task = new Task();
        FutureTask<String> futureTask = new FutureTask<>(task);
        executorService.submit(futureTask);
        String s = futureTask.get();
        System.out.println(s);
        executorService.shutdown();
    }
}

class Task implements Callable<String> {
    @Override
    public String call() throws Exception {
        // 模拟http
        System.out.println("http request");
        Thread.sleep(1000);

        return "request success";
    }
}

但我们可以使用 channel 配合 goroutine 实现类似的功能:

func main() {
 ch := Request("https://github.com")
 select {
 case r := <-ch:
  fmt.Println(r)
 }
}
func Request(url string) <-chan string {
 ch := make(chan string)
 go func() {
  // 模拟http请求
  time.Sleep(time.Second)
  ch <- fmt.Sprintf("url=%s, res=%s", url, "ok")
 }()
 return ch
}

goroutine 发起请求后直接将这个 channel 返回,调用方会在请求响应之前一直阻塞,直到 goroutine 拿到了响应结果。

goroutine 互相通信

   /**
     * 偶数线程
     */
    public static class OuNum implements Runnable {
        private TwoThreadWaitNotifySimple number;

        public OuNum(TwoThreadWaitNotifySimple number) {
            this.number = number;
        }

        @Override
        public void run() {
            for (int i = 0; i < 11; i++) {
                synchronized (TwoThreadWaitNotifySimple.class) {
                    if (number.flag) {
                        if (i % 2 == 0) {
                            System.out.println(Thread.currentThread().getName() + "+-+偶数" + i);

                            number.flag = false;
                            TwoThreadWaitNotifySimple.class.notify();
                        }

                    } else {
                        try {
                            TwoThreadWaitNotifySimple.class.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    }


    /**
     * 奇数线程
     */
    public static class JiNum implements Runnable {
        private TwoThreadWaitNotifySimple number;

        public JiNum(TwoThreadWaitNotifySimple number) {
            this.number = number;
        }

        @Override
        public void run() {
            for (int i = 0; i < 11; i++) {
                synchronized (TwoThreadWaitNotifySimple.class) {
                    if (!number.flag) {
                        if (i % 2 == 1) {
                            System.out.println(Thread.currentThread().getName() + "+-+奇数" + i);

                            number.flag = true;
                            TwoThreadWaitNotifySimple.class.notify();
                        }

                    } else {
                        try {
                            TwoThreadWaitNotifySimple.class.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    }

这里截取了”两个线程交替打印奇偶数“的部分代码。

Java 提供了 object.wait()/object.notify() 这样的等待通知机制,可以实现两个线程间通信。

go 通过 channel 也能实现相同效果:

func main() {
 ch := make(chan struct{})
 go func() {
  for i := 1; i < 11; i++ {
   ch <- struct{}{}
   //奇数
   if i%2 == 1 {
    fmt.Println("奇数:", i)
   }
  }
 }()

 go func() {
  for i := 1; i < 11; i++ {
   <-ch
   if i%2 == 0 {
    fmt.Println("偶数:", i)
   }
  }
 }()

 time.Sleep(10 * time.Second)
}

本质上他们都是利用了线程(goroutine)阻塞然后唤醒的特性,只是 Java 是通过 wait/notify 机制;

而 go 提供的 channel 也有类似的特性:

  1. channel 发送数据时(ch<-struct{}{})会被阻塞,直到 channel 被消费(<-ch)。

以上针对于无缓冲 channel

channel 本身是由 go 原生保证并发安全的,不用额外的同步措施,可以放心使用。

广播通知

不仅是两个 goroutine 之间通信,同样也能广播通知,类似于如下 Java 代码:

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    synchronized (NotifyAll.class){
                        NotifyAll.class.wait();
                    }
                    System.out.println(Thread.currentThread().getName() + "done....");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
        Thread.sleep(3000);
        synchronized (NotifyAll.class){
            NotifyAll.class.notifyAll();
        }
    }

主线程将所有等待的子线程全部唤醒,这个本质上也是通过 wait/notify 机制实现的,区别只是通知了所有等待的线程。

换做是 go 的实现:

func main() {
 notify := make(chan struct{})
 for i := 0; i < 10; i++ {
  go func(i int) {
   for {
    select {
    case <-notify:
     fmt.Println("done.......",i)
     return
    case <-time.After(1 * time.Second):
     fmt.Println("wait notify",i)

    }
   }
  }(i)
 }
 time.Sleep(1 * time.Second)
 close(notify)
 time.Sleep(3 * time.Second)
}

当关闭一个 channel 后,会使得所有获取 channelgoroutine 直接返回,不会阻塞,正是利用这一特性实现了广播通知所有 goroutine 的目的。

注意,同一个 channel 不能反复关闭,不然会出现panic。

channel 解耦

以上例子都是基于无缓冲的 channel,通常用于 goroutine 之间的同步;同时 channel 也具备缓冲的特性:

ch :=make(chan T, 100)

可以直接将其理解为队列,正是因为具有缓冲能力,所以我们可以将业务之间进行解耦,生产方只管往 channel 中丢数据,消费者只管将数据取出后做自己的业务。

同时也具有阻塞队列的特性:

  • channel 写满时生产者将会被阻塞。
  • channel 为空时消费者也会阻塞。

从上文的例子中可以看出,实现相同的功能 go 的写法会更加简单直接,相对的 Java 就会复杂许多(当然这也和这里使用的偏底层 api 有关)。

Java 中的 BlockingQueue

这些特性都与 Java 中的 BlockingQueue 非常类似,他们具有以下的相同点:

  • 可以通过两者来进行 goroutine/thread 通信。
  • 具备队列的特征,可以解耦业务。
  • 支持并发安全。

同样的他们又有很大的区别,从表现上看:

  • channel 支持 select 语法,对 channel 的管理更加简洁直观。
  • channel 支持关闭,不能向已关闭的 channel 发送消息。
  • channel 支持定义方向,在编译器的帮助下可以在语义上对行为的描述更加准确。

当然还有本质上的区别就是 channel 是 go 推荐的 CSP 模型的核心,具有编译器的支持,可以有很轻量的成本实现并发通信。

BlockingQueue 对于 Java 来说只是一个实现了并发安全的数据结构,即便不使用它也有其他的通信方式;只是他们都具有阻塞队列的特征,所有在初步接触 channel 时容易产生混淆。

相同点 channel 特有 阻塞策略 支持select 设置大小 支持关闭 并发安全 自定义方向 普通数据结构 编译器支持

总结

有过一门编程语言的使用经历在学习其他语言是确实是要方便许多,比如之前写过 Java 再看 Go 时就会发现许多类似之处,只是实现不同。

拿这里的并发通信来说,本质上是因为并发模型上的不同;

Go 更推荐使用通信来共享内存,而 Java 大部分场景都是使用共享内存来通信(这样就得加锁来同步)。

带着疑问来学习确实会事半功倍。

相关推荐

高性能并发队列Disruptor使用详解

基本概念Disruptor是一个高性能的异步处理框架,是一个轻量的Java消息服务JMS,能够在无锁的情况下实现队列的并发操作Disruptor使用环形数组实现了类似队列的功能,并且是一个有界队列....

Disruptor一个高性能队列_java高性能队列

Disruptor一个高性能队列前言说到队列比较熟悉的可能是ArrayBlockingQueue、LinkedBlockingQueue这两个有界队列,大多应用在线程池中使用能保证线程安全,但其安全性...

谈谈防御性编程_防御性策略

防御性编程对于程序员来说是一种良好的代码习惯,是为了保护自己的程序在不可未知的异常下,避免带来更大的破坏性崩溃,使得程序在错误发生时,依然能够云淡风轻的处理,但很多程序员入行很多年,写出的代码依然都是...

有人敲门,开水开了,电话响了,孩子哭了,你先顾谁?

前言哎呀,这种情况你肯定遇到过吧!正在家里忙活着,突然——咚咚咚有人敲门,咕噜咕噜开水开了,铃铃铃电话响了,哇哇哇孩子又哭了...我去,四件事一起来,人都懵了!你说先搞哪个?其实这跟我们写Java多线...

面试官:线程池如何按照core、max、queue的执行顺序去执行?

前言这是一个真实的面试题。前几天一个朋友在群里分享了他刚刚面试候选者时问的问题:"线程池如何按照core、max、queue的执行循序去执行?"。我们都知道线程池中代码执行顺序是:co...

深入剖析 Java 中线程池的多种实现方式

在当今高度并发的互联网软件开发领域,高效地管理和利用线程资源是提升程序性能的关键。Java作为一种广泛应用于后端开发的编程语言,为我们提供了丰富的线程池实现方式。今天,就让我们深入探讨Java中...

并发编程之《彻底搞懂Java线程》_java多线程并发解决方案详解

目录引言一、核心概念:线程是什么?...

Redis怎么实现延时消息_redis实现延时任务

一句话总结Redis可通过有序集合(ZSET)实现延时消息:将消息作为value,到期时间戳作为score存入ZSET。消费者轮询用ZRANGEBYSCORE获取到期消息,配合Lua脚本保证原子性获取...

CompletableFuture真的用对了吗?盘点它最容易被误用的5个场景

在Java并发编程中,CompletableFuture是处理异步任务的利器,但不少开发者在使用时踩过这些坑——线上服务突然雪崩、异常悄无声息消失、接口响应时间翻倍……本文结合真实案例,拆解5个最容易...

接口性能优化技巧,有点硬_接口性能瓶颈

背景我负责的系统到2021年初完成了功能上的建设,开始进入到推广阶段。随着推广的逐步深入,收到了很多好评的同时也收到了很多对性能的吐槽。刚刚收到吐槽的时候,我们的心情是这样的:...

禁止使用这5个Java类,每一个背后都有一段&quot;血泪史&quot;

某电商平台的支付系统突然报警:大量订单状态异常。排查日志发现,同一笔订单被重复支付了三次。事后复盘显示,罪魁祸首竟是一行看似无害的SimpleDateFormat代码。在Java开发中,这类因使用不安...

无锁队列Disruptor原理解析_无锁队列实现原理

队列比较队列...

Java并发队列与容器_java 并发队列

【前言:无论是大数据从业人员还是Java从业人员,掌握Java高并发和多线程是必备技能之一。本文主要阐述Java并发包下的阻塞队列和并发容器,其实研读过大数据相关技术如Spark、Storm等源码的,...

线程池工具及拒绝策略的使用_线程池处理策略

线程池的拒绝策略若线程池中的核心线程数被用完且阻塞队列已排满,则此时线程池的资源已耗尽,线程池将没有足够的线程资源执行新的任务。为了保证操作系统的安全,线程池将通过拒绝策略处理新添加的线程任务。...

【面试题精讲】ArrayBlockingQueue 和 LinkedBlockingQueue 区别?

有的时候博客内容会有变动,首发博客是最新的,其他博客地址可能会未同步,认准...