Java & Go 并发编程对比

1
2
3

Runnable task = ()-> System.out.println("task running");
Thread t = new Thread(task);
t.start();

1
2
3

Runnable task = ()-> System.out.println("task running");
Executor executor = Executors.newCachedThreadPool();
executor.execute(task);

1
2
3

go func() {
  fmt.Println("task running")
}()

1
2
3
4
5
6
7

public static void main(String[] args) throws InterruptedException {
  ScheduledExecutorService schExecutor = Executors.newSingleThreadScheduledExecutor();
  Runnable task = () -> System.out.println("tick  at " + LocalDateTime.now().getSecond());
  schExecutor.scheduleAtFixedRate(task, 1, 1  , TimeUnit.SECONDS);
  Thread.sleep(3000);
  schExecutor.shutdown();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func main() {
  go func() {
    ticker := time.Tick(time.Second)
    time.Sleep(time.Second)
    for {
      <-ticker
      fmt.Printf("tick at %d\n", time.Now().Second())
    }
  }()
  time.Sleep(3 * time.Second)
}

1
2
3

tick at 27
tick at 28
tick at 29

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

static CompletableFuture<String> asyncJob() {
    return CompletableFuture.supplyAsync(() -> {
        // some expensive job...
        return "finish";
    });
}

public static void main(String[] args) throws Exception {
  // 运行异步任务并注册回调
  asyncJob().whenComplete((ret, ex) -> {
      if (ex != null) {
          // 处理异常
      }
      // 处理运行结果
  });
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

type ResultErr struct {
  ret string
  err error
}

func asyncJob() <-chan ResultErr {
  retErr := make(chan ResultErr)
  go func() {
    // do expensive job
    // 如果中途发生异常，返回错误
    // retErr <- ResultErr{err: errors.New("error")}
    retErr <- ResultErr{ret: "finish"}
  }()
  return retErr
}

func main () {
  retErr := <-asyncJob()
  if retErr.err != nil {
    // 错误处理
  }
  // 结果处理
  // ret := retErr.ret
}

1
2
3
4
5
6
7

// ExecutorService

<T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
      throws InterruptedException;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

func invokeAny() {
  rets := make(chan struct{}, 10)
  stops := make(chan struct{}, 9)

  for i := 0; i < 10; i++ {
    go func() {
      select {
      case <-stops:
        return
      default:
        // execute job
        rets <- struct{}{}
      }
    }()
  }
  ret := <-rets
  fmt.Println(ret)

  for i := 0; i < 9; i++ {
    stops <- struct{}{}
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

private static final ThreadLocal<String> localMap = new ThreadLocal<>();
public static void main(String[] args) throws Exception {
  localMap.set("main hello");
  Thread t = new Thread(() -> {
      String tName = Thread.currentThread().getName();
      System.out.println(tName + " get " + localMap.get());
      localMap.set("sub hello");
      System.out.println(tName + " get " + localMap.get());
  });
  t.start();
  t.join();
  System.out.println("Thread-main get "+ localMap.get());
} //～
Thread-0 get null
Thread-0 get sub hello
Thread-main get main hello

1
2
3
4
5
6
7
8
9

func main() {
  ctx := context.WithValue(context.TODO(), "key", "value")
  withCtx := func(ctx context.Context, others ...int) {
    fmt.Println(ctx.Value("key"))
  }
  go withCtx(ctx, 1, 2, 3)
  time.Sleep(time.Millisecond)
} //~
value

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

public class CancelableTask {
  public static void main(String[] args) throws InterruptedException {
    interruptThread();
    interruptFuture();
  }

  static void interruptThread() throws InterruptedException {
    Thread t = new Thread(CancelableTask::loop);
    t.start();
    Thread.sleep(1);
    t.interrupt();
  }

  static void interruptFuture() throws InterruptedException {
    ExecutorService exeSvc = Executors.newSingleThreadExecutor();
    Future<?> ft = exeSvc.submit(CancelableTask::loop);
    Thread.sleep(1);
    ft.cancel(true);
    exeSvc.shutdown();
  }

  static void loop() {
    while (!Thread.interrupted()) {
        //... do business
        System.out.println("looping");
    }
    System.out.println("stopped");
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

func main() {
  stop := make(chan struct{})
  go loop(stop)
  <- time.After(time.Millisecond)
  close(stop) // 也可使用 stop <- struct{}{}
}

func loop(stop <-chan struct{}) {
  for {
    select {
    case <-stop:
      fmt.Println("stopped")
      return
    default:
      //do business
      fmt.Println("looping")
    }
  }
}

1
2
3
4

looping
...
looping
stopped

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

static void loopWithBlockQueue(BlockingQueue<Object> queue) {
    while (queue.poll() == null){
        //... do business
        System.out.println("looping");
    }
    System.out.println("stopped");
}

public static void main(String[] args) throws InterruptedException {
    BlockingQueue<Object> queue = new ArrayBlockingQueue<>(1);
    new Thread(() -> loopWithBlockQueue(queue)).start();
    Thread.sleep(1);
    queue.put(new Object());
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func main() {
  ctx, cancel := context.WithCancel(context.TODO())
  go func(ctx context.Context) {
    time.Sleep(time.Millisecond)
    select {
    case <-ctx.Done():
      fmt.Println("task canceled")
    default:
      fmt.Println("running")
      // do business
    }
  }(ctx)
  cancel()
  time.Sleep(2 * time.Millisecond)
}//~
task canceled

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

var wg = sync.WaitGroup{}
var stop = make(chan struct{})

func accept(task func()) error {
  select {
  case <-stop:
    return errors.New("reject")
  default:
    wg.Add(1)
    go func() {
      defer wg.Done()
      handle(task)
    }()
    return nil
  }
}

func handle(task func()) {
  task()
}

func shutdown() {
  close(stop)
  wg.Wait()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

public class Locker {
  public static void main(String[] args) {
      synchronized (Locker.class) {
          reentrant();
          synchronized (Lock.class){
              reentrant();
          }
      }
      System.out.println();
      ReentrantLock lock = new ReentrantLock();
      lock.lock();
      reentrant();
      lock.lock();
      reentrant();
      lock.unlock();
      lock.unlock();
  }

  static void reentrant() {
      System.out.println("entered");
  }
}//~
entered
entered

entered
entered

1
2
3

mu := sync.Mutex{}
mu.Lock() // ok
mu.Lock() // dead lock

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public interface Lock {
    void lock();
    
    void unlock();

    void lockInterruptibly() throws InterruptedException;

    boolean tryLock();

    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

public static void main(String[] args) throws InterruptedException {
  ReentrantLock lock = new ReentrantLock();
  lock.lock();
  Thread t = new Thread(() -> {
      try {
          // 中断式获取，被中断抛出 InterruptedException
          // lock.lockInterruptibly();
          // 超时获取，操作超时自动返回，被中断抛出 InterruptedException
          lock.tryLock(10, TimeUnit.MILLISECONDS);
      } catch (InterruptedException ex) {
          ex.printStackTrace();
      }
  });
  t.start();
  Thread.sleep(100);
  t.interrupt();
}//~
java.lang.InterruptedException
  at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireNanos(AbstractQueuedSynchronizer.java:992)
  at java.base/java.util.concurrent.locks.ReentrantLock$Sync.tryLockNanos(ReentrantLock.java:168)
  at java.base/java.util.concurrent.locks.ReentrantLock.tryLock(ReentrantLock.java:479)
  ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

var lockCh = make(chan struct{}, 1)

func TryLock(timeout time.Duration) bool {
  for {
    select {
    case lockCh <- struct{}{}:
      return true
    case <-time.After(timeout):
      return false
    }
  }
}

func LockInterruptibly(interrupt <-chan struct{}) bool {
   for {
    select {
    case <-interrupt:
      return false
    case lockCh <- struct{}{}:
      return true
    }
  }
}

func UnLock() {
  <-lockCh
}

1
2
3

Semaphore semaphore = new Semaphore(5);
semaphore.acquire(1);
semaphore.release(1);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

type Semaphore struct {
  permits chan struct{}
}

func NewSemaphore(permits uint) *Semaphore {
  permitsCh := make(chan struct{}, permits)
  return &Semaphore{permits: permitsCh}
}

func (s *Semaphore) Acquire() {
  s.permits <- struct{}{}
}

func (s *Semaphore) Release() {
  <-s.permits
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

func main() {
  var once = sync.Once{}
  f := func() {
    time.Sleep(10 * time.Millisecond)
    fmt.Println("do once")
  }
  go func() {
    fmt.Println("do once start")
    once.Do(f)
    fmt.Println("do once finish")
  }()
  time.Sleep(1 * time.Millisecond)
  for i := 0; i < 2; i++ {
    go func() {
      fmt.Println("block...")
      once.Do(f)
      fmt.Println("resume")
    }()
  }
  time.Sleep(10 * time.Millisecond)
}//~
do once start
block...
block...
do once
do once finish
resume
resume

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

func main () {
  wg := sync.WaitGroup{}
  wg.Add(3)

  for i := 0; i < 3; i++ {
    ii := i
    go func() {
      defer wg.Done()
      fmt.Printf("%d finished\n", ii)
    }()
  }

  wg.Wait()
  fmt.Println("all finish")
  // do exchange
} // ~
2 finished
0 finished
1 finished
all finish

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

var a string

func hello() {
  a = "hello"
  go func() { print(a) }()
}

func notHello(){
  go func() { a = "hello" }()
	print(a)
}