Python代码阅读（第61篇）：延迟调用函数

发表于 2021-11-23

Python 代码阅读合集介绍：为什么不推荐Python初学者直接看项目源码

本篇阅读的代码实现了在给定的延迟时间后，调用指定函数的功能。

本篇阅读的代码片段来自于30-seconds-of-python。

`delay`

python复制代码from time import sleep

def delay(fn, ms, *args):
  sleep(ms / 1000)
  return fn(*args)

# EXAMPLES
delay(
  lambda x: print(x),
  1000,
  'later'
) # prints 'later' after one second

delay函数接收一个指定函数fn，一个延迟时间ms和指定函数的参数*args，在指定延迟后，返回指定函数fn的调用结果。

函数使用sleep()方法来进行延迟，然后调用指定函数。delay函数在调用的时候，可以使用lambda表达式的匿名函数，也可以使用一般函数。需要注意的是当fn存在关键字参数时会发生异常。

python复制代码>>> from time import sleep
>>>
>>> def delay(fn, ms, *args):
...   sleep(ms / 1000)
...   return fn(*args)
...
>>> def f(pos_only, /, standard, *, kwd_only):
...     print('pos_only:{}\nstandard:{}\nkwd_only:{}'.format(pos_only,standard,kwd_only))
...
>>> delay(f, 1000, 1, 2, 3)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 3, in delay
TypeError: f() takes 2 positional arguments but 3 were given
>>> delay(f, 1000, 1, 2, kwd_only = 3)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: delay() got an unexpected keyword argument 'kwd_only'

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Rust内存和分配内存与分配结语

发表于 2021-11-23

这是我参与11月更文挑战的第23天，活动详情查看：2021最后一次更文挑战

内存与分配

上一篇文章最后留了一个小问题，为什么String类型可变而字面值却不行呢？区别就在于两种数据类型对内存的处理上。

就字符串字面值而言，在编译的时候就知道其内容是什么，因此文本被直接硬编码进最终的可执行文件中，这使得字符串字面值快速高效，但是这些特性都是基于字符串字面值的不可变性。开发人员如果想实现每一个在编译时大小未知的文本而将一块内存放入二进制文件中，并且它的大小还可能随着程序运行而改变这是无法实现的。

对于String类型而言，为了支持一个可变，可增长的文本片段，需要在堆上分配一块在编译时未知大小的内存来存放内容。这意味着：

在程序运行时需要向操作系统请求内存；
需要一个当我们处理完String时将内存返回给操作系统的方法

上面的第一部分由开发人员完成，当调用String::from时，就会向操作系统请求所需要的内存，这在编程语言中是非常通用的。

但是第二部分实现起来就各有区别了，在有 垃圾回收（garbage collector，GC）的语言中， GC 记录并清除不再使用的内存，所以开发人员并不需要关心它。但是如果没有 GC 的话，识别出不再使用的内存并调用代码显式释放就是开发人员的责任了，跟请求内存的时候一样。正确处理内存回收曾经是一个困难的编程问题，如果忘记回收了会浪费内存。如果过早回收了，将会出现无效变量。如果重复回收，这也是个 bug。我们需要精确的为一个 allocate 配对一个 free。

Rust采用了一个不同的策略，拥有某块内存的变量，当这个变量离开其作用域之后就会被自动释放。

rust复制代码fn main() {
    // i 在还没有声明的时候是无效的
    let i = "rust";  // 从此处起，i 是有效的
    
    println!("{}", i); // 可以对i变量进行操作
}
// i 作用域到此结束，i 不再有效

这是一个将String需要的内存返回给操作系统的很自然的位置，当i离开其作用域的时候，Rust 就会自动调用一个特殊的函数drop，该函数可以释放离开作用域的变量所占用的内存。

这种模式对编写Rust代码有深远的影响，虽然看起来简单，但是在更复杂的场景下代码的行为可能是不可预测的，比如当有多个变量使用在堆上分配的内存时。对于这些场景，作者将单独另写一篇文章哦~

结语

文章首发于微信公众号程序媛小庄，同步于掘金。

码字不易，转载请说明出处，走过路过的小伙伴们伸出可爱的小指头点个赞再走吧(╹▽╹)

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

用虚拟机模拟远程服务器-创建 Linux 服务器

发表于 2021-11-23

在软件开发中，无论软件是基于 B/S 架构还是 C/S 架构，其中的 S，也就是服务器 Server，都是不可或缺的一部分。但作为一名软件开发的初学者，想要搭建完整的 B/S 或 C/S 学习环境，基于云服务器的话，不促销的时候稍微有点昂贵，实体服务器就更不用说了。

那么是否有相对实惠一些的方案呢？以下是我用虚拟机模拟远程服务器的方案，使用到的工具都是免费的，希望能对大家有所帮助。

搭建工具

VMware Player：VMware 免费开放给个人用户在非商业场景下使用的虚拟化软件
Ubuntu Server：选择这款 Linux 发行版因为我目前只会用 apt 系的包管理工具……大家可以根据情况选择其它 Linux 发行版

搭建步骤

使用 VMware Player 创建虚拟机

打开 VMware Player 开始创建虚拟机

创建.png
2. 注意此处选择稍后安装，这样在之后处理不必要的设备时方便一点

稍后安装.png
3. 选择将要安装的 Linux 发行版

选择对应.png
4. 选择虚拟机存放的路径

存放路径.png
5. 选择虚拟磁盘的大小和策略，这里我选择的是 20G 和按单个文件存储

虚拟磁盘.png
6. 接下来创建程序会告知默认的设备配置，但其中一部分需要依据情况手动调整的

自定义硬件.png

* 为了模拟入门云服务器，我将内存设置为 2G，处理器设置为单核，并打开了虚拟性能增强


![性能.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/5c756149e8a347c318275e48451f75ed3f645c230a18f55c4002732bd9819b8e)
* 在 CD/DVD 设备中选中之前下载好的系统 iso 文件


![系统.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/4b24ec907fdf032c4a1a888c2d46b5926f466ff0d08e459d6d4cc2d529fe45b8)
* 移除暂时不必要的设备后就可以关闭设备配置了


![不必要.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/278a5d92478eae04a73e0170b9555258ee7918ec6c3126c5f6942d11a9650c6b)

虚拟机创建成功后，点击播放虚拟机开始操作系统的安装

虚拟机完成.png
8. 可选：VMware Player 默认以 legacy 模式启动虚拟机，但现在主流的启动方式为 UEFI 模式。如果想要开启 UEFI 模式，需要到虚拟机文件夹下，用文本编辑器打开 .vmx 文件并在文件末尾添加 firmware = "efi"

Ubuntu Server 安装

在自动初始化完成后，就进入了 Ubuntu Server 安装程序。在程序中可以通过方向键进行选择，通过回车键确认。

第一步是选择语言，我选择保持系统默认的 English

语言.png
2. 键盘布局也保持默认

键盘布局.png
3. 接下来是配置网络信息，默认情况下是从 DHCP 服务器自动获取，我们先使用获取到的动态 IP 地址，等系统安装完成后再作修改

网卡.png
4. 配置代理，由于我的虚拟机是通过 NAT 和宿主机共享 IP 地址实现和互联网连接，所以此处可以留空

代理.png
5. 配置软件仓库镜像源，因为很多时候网络会不太好，直接访问 Ubuntu 软件仓库会比较慢，所以需要配置网络环境更好的镜像源。这里我使用的是中国科学技术大学提供的镜像源

换源.png
6. 配置磁盘分区

* 因为我们使用的是虚拟磁盘，所以整个磁盘都可以给系统使用。除此之外我选择关闭暂时用不上的高级特性 LVM


![磁盘选择.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/f4f7f25288bc9b1380d78904d4c8261976e97c855e3cdf3f438f2a6a28a753f0)
* 接下来是对磁盘进行分区


![默认分区.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/950c4c64173a36260b26d9ea0c7585e7a928c611460f2e50c8b5a93e1b6be09b)


我们可以看到系统默认的分区方案为：



| 分区编号 | 大小 | 格式 | 挂载点 |
| --- | --- | --- | --- |
| 1 | 512M | FAT32 | `/boot/efi` |
| 2 | 剩余所有空间 | Ext4 | `/` |

即除了用于 UEFI 启动的 ESP 分区外，剩余所有存储空间都挂载在根目录。我选择保持该默认配置，大家可以根据自己的情况自定义。
* 接下来将开始正式安装系统，磁盘上的数据将被清除。如果是实机安装，大家一定要确认选中磁盘上没有重要数据


![确认磁盘.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/4b7dbd50506155f3e86e20367d953a7f07a3967d13df079c1f8696b221ca1f73)

创建登录账户

登录账户.png
8. 是否安装 OpenSSH，因为之后我们将通过 SSH 进行远程登录，所以此处选择安装软件包，但不导入认证密钥

9. 是否需要安装常用服务器软件包，我选择跳过该阶段，不选中任何软件包

10. 安装完成后我们选择重启

安装完成.png
11. 收尾工作

* 可以看到系统提示移除安装使用的 iso 文件


![提示移除.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/4f3718a8bbadff0ff16c4f6d067a0acd7d60d04256299eeeb53516c855b0babf)
* 打开虚拟机配置将 CD/DVD 设备移除，然后按回车键正式退出安装


![移除CD-1.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/52cbe5e4d95d6b3fef6420a957afdaa5e326b3bbadbd637bf433e5b2fd1ddd31)
![移除CD-2.png](https://gitee.com/songjianzaina/juejin_p12/raw/master/img/097536d919e4b5f2d8088030cda7b118554bf1efb511037639939e546154fd54)

再次启动虚拟机，输入先前创建账户对应的账号密码，我们就成功登录我们创建的服务器了！

正式登录.png

总结

在本篇文章中，我们通过使用 VMware Player 和 Ubuntu Server，一步步地完成了 Linux 服务器的安装。

长出一口气不是吗？但是我们的旅途还远没有结束，因为目前我们与服务器的互动方式更像是直接操作服务器实体，而在现实开发场景中，我们更多的是通过远程登录来与服务器互动。

在下一篇文章中，我们将从配置网卡开始，一步步地实现通过 SSH 远程登录服务器！

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Netty编程（四）—— Future与Promise

发表于 2021-11-23

概念

netty 中的 Future 与 jdk 中的 Future 同名，但是是两个接口，netty 的 Future 继承自 jdk 的 Future，而 Promise 又对 netty Future 进行了扩展（继承）

jdk Future 只能同步等待任务结束（或成功、或失败）才能得到结果
netty Future 可以同步等待任务结束得到结果，也可以异步方式得到结果，但都是要等任务结束，比如Netty编程（三）—— Channel - 掘金 (juejin.cn)说到的 addListener
netty Promise 不仅有 netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器

功能/名称	jdk Future	netty Future	Promise
cancel	取消任务	-	-
isCanceled	任务是否取消	-	-
isDone	任务是否完成，不能区分成功失败	-	-
get	获取任务结果，阻塞等待	-	-
getNow	-	获取任务结果，非阻塞，还未产生结果时返回 null	-
await	-	等待任务结束，如果任务失败，不会抛异常，而是通过 isSuccess 判断	-
sync	-	等待任务结束，如果任务失败，抛出异常	-
isSuccess	-	判断任务是否成功	-
cause	-	获取失败信息，非阻塞，如果没有失败，返回null	-
addLinstener	-	添加回调，异步接收结果	-
setSuccess	-	-	设置成功结果
setFailure	-	-	设置失败结果

JDK Future

java复制代码public class JdkFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ThreadFactory factory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "JdkFuture");
            }
        };
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10,10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), factory);

        // 获得Future对象
        Future<Integer> future = executor.submit(new Callable<Integer>() {

            @Override
            public Integer call() throws Exception {
                TimeUnit.SECONDS.sleep(1);
                return 50;
            }
        });

        // 通过阻塞的方式，获得运行结果
        System.out.println(future.get());
    }
}

Netty Future

java复制代码public class NettyFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();

        // 获得 EventLoop 对象
        EventLoop eventLoop = group.next();
        Future<Integer> future = eventLoop.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 50;
            }
        });

        // 主线程中获取结果
        System.out.println(Thread.currentThread().getName() + " 获取结果");
        System.out.println("getNow " + future.getNow());
        System.out.println("get " + future.get());

        // NIO线程中异步获取结果
        future.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                System.out.println(Thread.currentThread().getName() + " 获取结果");
                System.out.println("getNow " + future.getNow());
            }
        });
    }
}

运行结果

csharp复制代码main 获取结果
getNow null
get 50
nioEventLoopGroup-2-1 获取结果
getNow 50

Netty中的Future对象，可以通过EventLoop的sumbit()方法得到

可以通过Future对象的get方法，阻塞地获取返回结果
也可以通过getNow方法，获取结果，若还没有结果，则返回null，该方法是非阻塞的
还可以通过future.addListener方法，在Callable方法执行的线程中，异步获取返回结果

Netty Promise

Promise相当于一个容器，可以用于存放各个线程中的结果，然后让其他线程去获取该结果

java复制代码public class NettyPromise {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建EventLoop
        NioEventLoopGroup group = new NioEventLoopGroup();
        EventLoop eventLoop = group.next();

        // 创建Promise对象，结果容器，用于存放结果
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 自定义线程向Promise中存放结果
            promise.setSuccess(50);
        }).start();

        // 主线程从Promise中获取结果
        System.out.println(Thread.currentThread().getName() + " " + promise.get());
    }
}

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

go channel 关闭的那些事儿 go channel

发表于 2021-11-23

go channel 关闭的那些事儿

什么情况下关闭 channel 会造成 panic？有没有必要关闭 channel？如何判断 channel 是否关闭？如何优雅地关闭 channel？这些你都知道吗？（不要告诉我你只会回答最后一个问题！）看到这一溜烟的问题，不知道你会不会不禁感叹，究竟哪个天杀的总说 go channel “哲学”“优雅”的？也许 Rob Pike （go语言之父）会说：嗯，当然“哲学”“优雅”，只是需要你注意的问题有点多……

什么情况下关闭 channel 会造成 panic ？

先看示例：

go复制代码 // 1.未初始化时关闭
 func TestCloseNilChan(t *testing.T) {
    var errCh chan error
    close(errCh)
    
    // Output:
    // panic: close of nil channel
 }
 
 // 2.重复关闭
 func TestRepeatClosingChan(t *testing.T) {
    errCh := make(chan error)
    var wg sync.WaitGroup
    wg.Add(1)
 
    go func() {
       defer wg.Done()
       close(errCh)
       close(errCh)
    }()
 
    wg.Wait()
    
    // Output:
    // panic: close of closed channel
 }
 
 // 3.关闭后发送
 func TestSendOnClosingChan(t *testing.T) {
    errCh := make(chan error)
    var wg sync.WaitGroup
    wg.Add(1)
 
    go func() {
       defer wg.Done()
       close(errCh)
       errCh <- errors.New("chan error")
    }()
 
    wg.Wait()
    
    // Output:
    // panic: send on closed channel
 }
 
 // 4.发送时关闭
 func TestCloseOnSendingToChan(t *testing.T) {
    errCh := make(chan error)
    var wg sync.WaitGroup
    wg.Add(1)
 
    go func() {
       defer wg.Done()
       defer close(errCh)
 
       go func() {
          errCh <- errors.New("chan error") // 由于 chan 没有缓冲队列，代码会一直在此处阻塞
       }()
 
       time.Sleep(time.Second) // 等待向 errCh 发送数据
    }()
 
    wg.Wait()
 
    // Output:
    // panic: send on closed channel
 }

综上，我们可以总结出如下知识点：

【知识点】在下述 4 种情况关闭 channel 会引发 panic：未初始化时关闭、重复关闭、关闭后发送、发送时关闭。

另外，从 golang 的报错中我们可以知道，golang 认为第3种和第4种情况属于一种情况。

通过观察上述代码，为避免在使用 channel 时遇到重复关闭、关闭后发送的问题，我想我们可以总结出以下两点规律：

应该只在发送端关闭 channel。（防止关闭后继续发送）
存在多个发送者时不要关闭发送者 channel，而是使用专门的 stop channel。（因为多个发送者都在发送，且不可能同时关闭多个发送者，否则会造成重复关闭。发送者和接收者多对一时，接收者关闭 stop channel；多对多时，由任意一方关闭 stop channel，双方监听 stop channel 终止后及时停止发送和接收）

这两点规律被称为“channel 关闭守则”。

既然关闭 channel 这么麻烦，那么我们有没有必要关闭 channel 呢？不关闭又如何？

有没有必要关闭 channel？不关闭又如何？

我们考虑以下两种情况：

情况一：channel 的发送和接收次数确定且相同时

go复制代码 func TestIsCloseChannelNecessary_on_equal(t *testing.T) {
     fmt.Println("NumGoroutine:", runtime.NumGoroutine())
     ich := make(chan int)
 
     // sender
     go func() {
        for i := 0; i < 3; i++ {
           ich <- i
        }
     }()
 
     // receiver
     go func() {
        for i := 0; i < 3; i++ {
           fmt.Println(<-ich)
        }
     }()
 
     time.Sleep(time.Second)
     fmt.Println("NumGoroutine:", runtime.NumGoroutine())
    
     // Output:
     // NumGoroutine: 2
     // 0
     // 1
     // 2
     // NumGoroutine: 2
 }

channel 的发送和接收次数确定且相同时，发送者 go routine 和接收者 go routine 分别都会在发送或接收结束时结束各自的 go routine。而上述代码中的 ich 会由于没有代码使用被垃圾收集器回收。因此这种情况下，不关闭 channel，没有任何副作用。

情况二：channel 的发送次数不确定时

go复制代码 func TestIsCloseChannelNecessary_on_less_sender(t *testing.T) {
    fmt.Println("NumGoroutine:", runtime.NumGoroutine())
    ich := make(chan int)
 
    // sender
    go func() {
       for i := 0; i < 2; i++ {
          ich <- i
       }
    }()
 
    // receiver
    go func() {
       for i := 0; i < 3; i++ {
          fmt.Println(<-ich)
       }
    }()
 
    time.Sleep(time.Second)
    fmt.Println("NumGoroutine:", runtime.NumGoroutine())
    
    // Output:
    // NumGoroutine: 2
    // 0
    // 1
    // NumGoroutine: 3
 }

以上述代码为例，channel 的发送次数小于接收次数时，接收者 go routine 由于等待发送者发送一直阻塞。因此接收者 go routine 一直未退出，ich 也由于一直被接收者使用无法被垃圾回收。未退出的 go routine 和未被回收的 channel 都造成了内存泄漏的问题。

因此，在发送者与接收者一对一的情况下，只要我们确保发送者或接收者不会阻塞，不关闭 channel 是可行的。在我们无法准确判断 channel 的发送次数和接收次数时，我们应该在合适的时机关闭 channel。那么如何判断 channel 是否关闭呢？

如何判断 channel 是否关闭？

【知识点】go channel 关闭后，读取该 channel 永远不会阻塞，且只会输出对应类型的零值。

如下代码所示：

go复制代码 func TestReadFromClosedChan(t *testing.T) {
    var errCh = make(chan error)
 
    go func() {
       defer close(errCh)
       errCh <- errors.New("chan error")
    }()
 
    go func() {
       for i := 0; i < 3; i++ {
          fmt.Println(i, <-errCh)
       }
    }()
 
    time.Sleep(time.Second)
    
    // Output:
    // 0 chan error
    // 1 <nil>
    // 2 <nil>
 }

以上述代码为例，nil 可能也是需要 channel传输的值之一，通常我们无法通过判断是否为类型的零值确定 channel 是否关闭。所以为了避免输出无意义的值，我们需要一种合理的方式判断 channel 是否关闭。golang 官方为我们提供了两种方式。

解决方案一：使用 channel 的多重返回值（如 err, ok := <-errCh ）

go复制代码 func TestReadFromClosedChan2(t *testing.T) {
     var errCh = make(chan error)
     go func() {
         defer close(errCh)
         errCh <- errors.New("chan error")
     }()
 
     go func() {
         for i := 0; i < 3; i++ {
             err, ok := <-errCh
             if ok {
                 fmt.Println(i, err)
             } else {
                 fmt.Println(i, err)
             }
 
         }
     }()
 
     time.Sleep(time.Second)
 
     // Output:
     // 0 chan error
     // 1 <nil>
     // 2 <nil>
 }

err, ok := <-errCh 的第二个返回值 ok 表示 errCh 是否已经关闭。如果已关闭，则返回 false。

解决方案二：使用 for range 简化语法

go复制代码 func TestReadFromClosedChan(t *testing.T) {
    var errCh = make(chan error)
    go func() {
       defer close(errCh)
       errCh <- errors.New("chan error")
    }()
 
    go func() {
       i := 0
       for err := range errCh {
          fmt.Println(i, err)
          i++
       }
    }()
 
    time.Sleep(time.Second)
    
    // Output:
    // 0 chan error
 }

for range 语法会自动判断 channel 是否结束，如果结束则自动退出 for 循环。

如何优雅地关闭 channel ？

我们从前文也了解到，如果发生重复关闭、关闭后发送等问题，会造成 channel panic。那么如何优雅地关闭 channel，是我们关心的一个问题。

golang 官方为我们提供了一种方式，可以用来尽量避免这个问题。golang 允许我们在把 channel 作为参数时，使用 <- 控制 channel 发送方向，防止我们在错误的时候关闭 channel。

go复制代码 func TestOneSenderOneReceiver(t *testing.T) {
    ich := make(chan int)
    go sender(ich)
    go receiver(ich)
 }
 
 func sender(ich chan<- int) { // 注意参数中的箭头
    for i := 0; i < 100; i++ {
       ich <- i
    }
 }
 
 func receiver(ich <-chan int) {  // 注意参数中的箭头
    fmt.Println(<-ich)
    close(ich) // 此处代码会在编译期报错
 }

使用这种方法时，由于 close() 函数只能接受 chan<- T 类型的 channel，如果我们尝试在接收方关闭 channel，编译器会报错，所以我们可以在编译期提前发现错误。

除此之外，我们也可以使用如下的结构体（抄自go101《如何优雅地关闭 go channels》，做了一点修改，链接为此文的中文翻译）：

go复制代码 type Channel struct {
    C      chan interface{}
    closed bool
    mut    sync.Mutex
 }
 
 func NewChannel() *Channel {
    return NewChannelSize(0)
 }
 
 func NewChannelSize(size int) *Channel {
    return &Channel{
       C:      make(chan interface{}, size),
       closed: false,
       mut:    sync.Mutex{},
    }
 }
 
 func (c *Channel) Close() {
    c.mut.Lock()
    defer c.mut.Unlock()
    if !c.closed {
       close(c.C)
       c.closed = true
    }
 }
 
 func (c *Channel) IsClosed() bool {
    c.mut.Lock()
    defer c.mut.Unlock()
    return c.closed
 }
 
 func TestChannel(t *testing.T) {
    ch := NewChannel()
    println(ch.IsClosed())
    ch.Close()
    ch.Close()
    println(ch.IsClosed())
 }

该方案可以解决重复关闭锁的问题以及锁是否关闭的问题。通过 Channel.IsClosed() 判断是否关闭 channel ，又可以安全地发送和接收。当然我们也可以把 sync.Mutex 换成 sync.Once，来只让 channel 关闭一次。具体可以参考《如何优雅地关闭 go channels》。

有时候我们的代码已经使用了原生的 chan，或者我们不想使用单独的数据结构，也可以使用下述的几种方案。通常情况下，我们只会遇到四种需要关闭 channel 的情况（以下内容是我对《如何优雅地关闭 go channels》中方法的总结）：

一个发送者，一个接收者：发送者关闭 channel，接收者使用 select 或 for range 判断 channel 是否关闭。
一个发送者，多个接收者：发送者关闭 channel，同上。
多个发送者，一个接收者：接收者接收完毕后，使用专用的 stop channel 关闭；发送者使用 select 监听 stop channel 是否关闭。
多个发送者，多个接收者：任意一方使用专用的 stop channel 关闭；发送者、接收者都使用 select 监听 stop channel 是否关闭。

因此我们只需要熟记面对这四种情况时如何关闭 channel 即可。

总述

代码不会撒谎。事实证明，使用 go channel 要注意的问题确实不少。如果在关闭 channel 时处理不当，可能会导致 panic，甚至还会造成内存泄漏。关于内存泄漏的问题，我在文章《老手也常误用！详解 Go channel 内存泄漏问题》中详细进行了阐述，同时也对如何更好地关闭 channel 进行了详细介绍。有兴趣可以点击链接阅读。

另外，欢迎关注同名公众号：柳双六

历史文章

你的 golang 程序正在悄悄内存泄漏

什么是redo log？redo log 如何保证数据库不丢数据的？（MySQL两阶段提交详解）

MySQL 数据库行级锁的那些事儿！记录锁、间隙锁、临键锁和加锁规则

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

网络协议基础学习(三) MAC地址、IP地址、子网掩码、超

发表于 2021-11-23

一、MAC地址

每一个网卡, 都有一个6字节(48bit)的MAC地址, 全球唯一, 固化在了那个卡的ROM中, 由IEEE802标准规定
- 前三个字节: 组织唯一标识符, 有IEEE的注册管理机构分配给厂商
- 后三个字节: 网络接口标识符, 由厂商自己决定
以诺基亚的一个网卡为例

组织唯一标识符	网络接口标识符
40-55-82	0A-8C-6D

OUI查询
- standards-oui.ieee.org/oui/oui.txt
- mac.bmcx.com/

1、MAC地址的表现形式

Windows: 每个字节之间用横线(-)间隔
- 40-55-82-0A-8C-6D
Linux、Android、Mac、iOS: 每个字节之间用冒号(:)间隔
- 40:55:82:0A:8C:6D
Packet Tracer: 每两个字节间用点(.)间隔
- 4055.820A.8C6D
当48位全为1时, 代表广播地址
Windows电脑可以在终端使用命令ipconfig /all查看网卡的MAC地址
Mac电脑可以在终端使用命令ifconfig -a查看网卡地址的MAC地址

2、MAC地址的获取

当不知道对方主机的MAC地址时, 可以通过发送ARP广播获取对方的MAC地址

获取成功后, 会缓存IP地址、MAC地址的映射信息, 俗称ARP缓存
通过ARP广播获取的MAC地址, 属于动态(dynamic)缓存
- 存储时间比较短(默认是2分钟), 过期了就自动删除, 因为对方的MAC地址是可以更改的, 一直使用缓存会找不到目标主机

相关命令
- arp -a: 查询所有的arp缓存
- arp -a [主机地址]: 查询arp缓存
- arp -d [主机地址]: 删除arp缓存
- arp -s 主机地址 MAC地址: 增加一条缓存信息(静态缓存, 存储时间较久, 不同系统的存储时间不同)

二、ARP

ARP: 地址解析协议, 通过IP地址获取MAC地址
RARP: 逆地址解析协议
- 使用与ARP相同的报头结构
- 作用于ARP相反, 用于将MAC地址转换为IP地址
- 后来被BOOTP、DHCP所取代

三、ICMP

ICMP: 互联网控制消息协议
- IPv4中的ICMP被称作ICMPv4, IPv6中的ICMP则被称作ICMPv6
- 通常用于返回错误信息
  - 比如TTL过期值, 目的不可达
- ICMP的错误信息总是包括了源数据并返回给发送者

四、IP地址

IP地址: 互联网上的每一个主机, 都有一个IP地址
- 最初版是IPv4版本, 32bit(4字节), 2019年11月25日, 全球的IP地址已经用完
- 后面推出了IPv6版本, 128bit(16字节)

注意: 这里不指定版本的情况下, 默认就是IPv4

以IP地址192.168.1.10为例

1100 0000	1010 1000	0000 0001	0000 1010
第1部分	第2部分	第3部分	第4部分
1100 0000	1010 1000	0000 0001	0000 1010
192	168	1	10

1、IP地址的组成

IP地址由两部分组成: 网络标识(网络ID)、主机标识(主机ID)
- 同一个网段的计算机, 网络ID相同
- 通过子网掩码, 可以计算出网络ID: 子网掩码 & IP地址

yaml复制代码  1100 0000.1010 1000.0000 0001.0000 1010      IP地址: 192.168.1.10
& 1111 1111.1111.1111.1111 1111.0000 0000      子网掩码: 255.255.255.0
------------------------------------------
  1100 0000.1010 1000.0000 0001.0000 0000      网段: 192.168.1.0
  
其中与子网掩码 1111 1111.1111.1111.1111 1111 对应的 1100 0000.1010 1000.0000 0001 就是网络ID
子网掩码 0000 0000 对应的 0000 1010 就是主机ID

网段: 除网络ID外, 主机ID为0的地址, 例如 192.168.1.0
只有IP地址是没有办法得到网段的, 例如IP地址130.168.1.10, 我们看不出这个IP地址的网段是什么
必须有子网掩码才能算出IP地址所在网段, 例如子网掩码255.255.0.0, 可以算出IP地址130.168.1.10的网段是130.168.0.0, 其中网络ID130.168, 主机ID0.0
IP地址中的主机标识是分配给同一网段不同计算机的
- 例如: 网段130.168.0.0, 理论上可以分配给130.168.0.0~130.168.255.255, 共256 * 256 = 65536台计算机
- 但是, 130.168.0.0代表网段, 不能分配给计算机
- 130.168.255.255代表广播地址, 即: 给同一网段内的所有相连计算机发送广播
- 所以, 计算机数量是 256 * 256 - 2 = 65534台计算机
Packet Tracer上放置四台计算机, 它们的子网掩码统一为255.255.255.0, IP地址通过笔记标注, 所以它们都在同一个网段192.168.1.0上

通过ping 192.168.1.255, 可以给所有的计算机发送广播, 既: 主机ID的最大值255是用来给同一网段所有计算机发送广播的, 不能分配给某台主机当做主机ID

所以, 主机ID的最小是0和最大值255, 都有特殊用途, 不能分配给主机作为主机ID使用
例如: 网段130.168.0.0

+ 其中的主机位最小值`130.168.0.0`用来作为网段,
+ 主机位最大值`130.168.255.255`作为广播地址
+ 所以可分配的计算机IP范围是`130.168.0.1`~`130.168.255.254`, 数量是`256 * 256 - 2 = 65534`

计算机和其它计算机通信前, 会先判断目标主机和自己是否在同一网段

+ 同一网段: 不需要由路由器进行转发
+ 不同网段, 交由路由器进行转发

2、IP地址的分类

A类地址: 默认子网掩码 255.0.0.0

8bit 24bit

网络ID

0开头 ID主机
B类地址: 默认子网掩码 255.255.0.0

16bit 16bit

网络ID

10开头 ID主机
C类地址: 默认子网掩码 255.255.255.0

24bit 8bit

网络ID

110开头 ID主机
D类地址: 以1110开头, 多播地址
E类地址: 以1111开头, 保留为今后使用
只有A/B/C类地址才能分配给主机使用

8bit	24bit
网络ID
`0`开头	ID主机

16bit	16bit
网络ID
`10`开头	ID主机

24bit	8bit
网络ID
`110`开头	ID主机

(1) A类地址

因为网络ID第一位必须是0, 所以取值范围值0000 0000 ~ 0111 1111

网络ID
- 0不能用, 127作为保留字段, 其中127.0.0.1是本地回环地址, 代表主机地址
- 可以分配给主机的第1部分取值范围是1~126
主机ID
- 第2、3、4部分的取值是0~255
- 其中0.0.0和255.255.255分别是网段和广播地址, 不可以分配
- 所以, 每个A类网络能容下的最大主机数是 256 * 256 * 256 - 2 = 2^24 - 2 = 16777214

(2) B类地址

因为网络ID前两位必须是10, 所以取值范围值1000 0000 0000 0000 ~ 1011 1111 1111 1111

网络ID
- 可以分配给主机的第一部分取值范围是128~191, 第二部分的取值范围是0~255
主机ID
- 第3、4部分的取值范围是0~255
- 每个B类网络能容纳的最大主机数是: 256 * 256 - 2 = 2^16 - 2 = 65534

(3) C类地址

因为网络ID前三位必须是110, 所以取值范围值1100 0000 0000 0000 0000 0000 ~ 1101 1111 1111 1111 1111 1111

网络ID
- 可以分配给主机的
- 第1部分的取值范围是 192~223
- 第2、3部分的取值范围是 0~255
主机ID
- 第4部分的取值范围是 0~255
- 每个C类地址能够容纳的最大主机数是 256 - 2 = 254

(4) D类地址

没有子网掩码, 用于多播(组播)地址
- 第1部分取值范围是 224~239

(5) E类地址

保留为今后使用
- 第一部分取值范围是 240~255

三、子网掩码

1、子网掩码的CIDR表示方法

CIDR: 无类别域间路由
子网掩码的CIDR表示方法
- 192.168.1.100/24, 代表子网掩码有24个1, 也就是 255.255.255.0
- 123.210.100.200/16, 代表子网掩码有16个1, 也就是 255.255.0.0
以/将IP地址和子网掩码划分, 例如:
- 192.168.1.100/24中IP地址是192.168.1.100, 子网掩码是24个1, 即: 1111 1111 1111 1111 1111 1111 0000 0000, 等价于255.255.255.0
- 123.210.100.200/16中IP地址是123.210.100.200, 子网掩码是16个1, 即: 1111 1111 1111 1111 0000 0000 0000 0000, 等价于255.255.0.0

2、为什么要进行子网划分?

如果需要让200台主机在同一个网段内, 可以分配一个C类网段, 比如 192.168.1.0/24
- 共254个可用IP地址: 192.168.1.1 ~ 192.168.1.254
- 多出54个空闲的IP地址, 这种情况并不算浪费资源
如果需要让500台主机在同一个网段内, 那就分配一个B类网段, 比如 191.100.0.0/16
- 共65534个可用IP地址: 191.100.0.1 ~ 191.100.255.254
- 多出65034个空闲的IP地址, 这种情况属于极大的浪费资源

如何尽量免IP地址浪费?

合理进行子网划分。

3、子网划分

子网划分: 借用主机位作为子网位, 划分出多个子网
可用划分

+ 等长子网划分: 将一个网段等分成多个子网, 每个字网的可用IP地址数量是一样的
+ 变长子网划分: 每个字网的可用IP地址数量可以使不一样的

子网划分器: www.ab126.com/web/3552.ht…
子网的划分步骤

+ 确定子网的子网掩码长度
+ 确定子网中第1个、最后1个主机可用的IP地址

4、等长子网划分 - 等分成2个子网

现有一C类地址, 网络ID是 192.168.0, 子网掩码是 255.255.255.0, 即网段是 192.168.0.0/24, 一共可配置254台主机
现将该网段划分成两个字网, 只需要将子网掩码右移一位即可, 即: 255.255.255.128/25
A网段:
- 主机ID范围是192.168.0.1 ~ 192.168.0.126
- 网段: 192.168.0.0
- 广播地址: 192.168.0.127

yaml复制代码  1100 0000.1010 1000.0000 0000.0000 0000
& 1111 1111.1111 1111.1111 1111.1000 0000   子网掩码: 255.255.255.128/25
-----------------------------------------
  1100 0000.1010 1000.0000 0000.0000 0000   网段: 192.168.0.0/25

B网段
- 主机ID范围只192.168.0.129 ~ 192.168.0.254
- 网段: 192.168.0.128
- 广播地址: 192.168.0.255

yaml复制代码  1100 0000.1010 1000.0000 0000.1000 0000
& 1111 1111.1111 1111.1111 1111.1000 0000   子网掩码: 255.255.255.128/25
-----------------------------------------
  1100 0000.1010 1000.0000 0000.1000 0000   网段: 192.168.0.128/25

5、等长子网划分 - 等分成4个子网

网段192.168.0.0/24, 划分成四个子网, 只需要将子网掩码右移两位

1 2	yaml复制代码子网掩码: 1111 1111.1111 1111.1111.1111.0000 0000 255.255.255.0/24 右移两位: 1111 1111.1111 1111.1111.1111.1100 0000 255.255.255.192/26

A网段: 192.168.0.0/26

yaml复制代码  1100 0000.1010 1000.0000 0000.0000 0000
& 1111 1111.1111 1111.1111 1111.1100 0000
------------------------------------------
  1100 0000.1010 1000.0000 0000.0000 0000

B网段: 192.168.0.64/26

yaml复制代码  1100 0000.1010 1000.0000 0000.0100 0000
& 1111 1111.1111 1111.1111 1111.1100 0000
------------------------------------------
  1100 0000.1010 1000.0000 0000.0100 0000

C网段: 192.168.0.128/26

yaml复制代码  1100 0000.1010 1000.0000 0000.1000 0000
& 1111 1111.1111 1111.1111 1111.1100 0000
------------------------------------------
  1100 0000.1010 1000.0000 0000.1000 0000

D网段: 192.168.0.192/26

yaml复制代码  1100 0000.1010 1000.0000 0000.1100 0000
& 1111 1111.1111 1111.1111 1111.1100 0000
------------------------------------------
  1100 0000.1010 1000.0000 0000.1100 0000

6、变长子网划分

如果一个子网地址块的长度是原网段的(1/2)^n, 那么
- 子网的子网掩码, 就是在原网段的子网掩码基础上增加n个1
- 不等长的子网, 它们的子网掩码也不同

假设上图是对192.168.0.0/24进行变长子网划分
- C网段: 子网掩码是255.255.255.128/25. 即: 1111 1111.1111 1111.1111 1111.1000 0000
- B网段: 子网掩码是255.255.255.192/26. 即: 1111 1111.1111 1111.1111 1111.1100 0000
- A网段: 子网掩码是255.255.255.192/27. 即: 1111 1111.1111 1111.1111 1111.1110 0000
- D网段: 子网掩码是255.255.255.192/30. 即: 1111 1111.1111 1111.1111 1111.1111 1100
- E网段: 子网掩码是255.255.255.192/30. 即: 1111 1111.1111 1111.1111 1111.1111 1100

7、思考题

下面两台设备能正常通信吗?

第一台设备
- IP地址: 192.168.0.10
- 子网掩码: 255.255.255.0
- 网段: 192.168.0.0
第二台设备
- IP地址: 192.168.10.10
- 子网掩码: 255.255.0.0
- 网段: 192.168.0.0
可以发现, 两台设备的网段相同, 理论上是可以发送信息的

而实际上, 计算机0是无法向计算机1发送信息的
这是因为, 计算机0不知道计算机1的子网掩码, 只知道计算机1的IP地址, 所以在计算计算机1的网段时, 使用的是计算机0本身的子网掩码
- 计算机0的子网掩码: 255.255.255.0
- 计算机1的IP地址: 192.168.10.10
- 计算后, 计算机1的网段时: 192.168.10.0
- 所以, 计算机0无法给计算机1发送信息

四、超网

超网: 跟子网反过来, 它是将多个连续的网段合并成一个更大的网段
需求: 原本有200台计算机使用192.168..0.0/24网段, 现在希望增加200台设备到同一个网段
- 200台在192.168.0.0/24网段, 200台在192.168.1.0/24网段
- 合并192.168.0.0/24、192.168.1.0/24为一个网段: 192.168.0.0/23(子网掩码往左移动以为)

yaml复制代码192.168.0.0/24
  1100 0000.1010 1000.0000 0000.0000 0000
& 1111 1111.1111 1111.1111 1111.0000 0000
-----------------------------------------
  1100 0000.1010 1000.0000 0000.0000 0000

yaml复制代码192.168.1.0/24
  1100 0000.1010 1000.0000 0001.0000 0000
& 1111 1111.1111 1111.1111 1111.0000 0000
-----------------------------------------
  1100 0000.1010 1000.0000 0001.0000 0000

yaml复制代码192.168.0.0/23
  1100 0000.1010 1000.0000 0000.0000 0000
& 1111 1111.1111 1111.1111 1110.0000 0000
-----------------------------------------
  1100 0000.1010 1000.0000 0000.0000 0000

1、思考

192.168.0.255/23这个IP地址, 可以分配给计算机使用么?

yaml复制代码192.168.0.255/23
  1100 0000.1010 1000.0000 000|0.1111 1111       IP地址: 192.168.0.255
& 1111 1111.1111 1111.1111 111|0.0000 0000       子网掩码: 255.255.254.0
------------------------------|-----------
  1100 0000.1010 1000.0000 000|0.0000 0000       网段: 192.168.0.0/23

由此可知, 192.168.0.255/23 并不是网段192.168.0.0/23的最大IP地址, 即不是广播地址
所以, 192.168.0.255/23这个IP地址可以分配给计算机使用
192.168.0.0/23的最大IP地址是 192.168.0.1/23, 即: 1100 0000.1010 1000.0000 0001.1111 1111

2、合并4个网段

子网掩码向左移动2位, 可以合并4个网段

将192.168.0.0/24、192.168.1.0/24、192.168.2.0/24、192.168.3.0/24合并为192.168.0.0/22网段

3、思考

下面的两个网段, 能通过子网掩码左移1位进行合并么?

192.168.1.0/24子网掩码左移一位, 网段是192.168.0.0/23

yaml复制代码192.168.0.0/23
  1100 0000.1010 1000.0000 0001.0000 0000     192.168.1.0/24
& 1111 1111.1111 1111.1111 1110.0000 0000     255.255.254.0
-----------------------------------------
  1100 0000.1010 1000.0000 0000.0000 0000     192.168.0.0/23

192.168.2.0/24子网掩码左移一位, 网段是192.168.2.0/23

yaml复制代码192.168.2.0/23
  1100 0000.1010 1000.0000 0010.0000 0000     192.168.1.0/24
& 1111 1111.1111 1111.1111 1110.0000 0000     255.255.254.0
-----------------------------------------
  1100 0000.1010 1000.0000 0010.0000 0000     192.168.2.0/23

左移一位后, 192.168.0.0/23和192.168.2.0/23不在同一个网段, 所以不能合并为同一个网段

4、合并网段的规律

(1)子网掩码左移, 可以合并的网段数

合并两个网段, 子网掩码左移1位, 合并四个网段, 子网掩码左移2位, 合并八个网段, 子网掩码左移3位
即: 假设n为网段数, n是2的k次幂(k>=1), 子网掩码左移k位能够合并n个网段

(2)可以合并的网段

假设n是2的k次幂(k>=1), 现有下列网段

yaml复制代码1100 0000.1010 1000.0000 0000.0000 0000    192.168.0.0/24 
1100 0000.1010 1000.0000 0001.0000 0000    192.168.1.0/24 
1100 0000.1010 1000.0000 0010.0000 0000    192.168.2.0/24 
1100 0000.1010 1000.0000 0011.0000 0000    192.168.3.0/24 
1100 0000.1010 1000.0000 0100.0000 0000    192.168.4.0/24 
1100 0000.1010 1000.0000 0101.0000 0000    192.168.5.0/24 
1100 0000.1010 1000.0000 0110.0000 0000    192.168.6.0/24 
1100 0000.1010 1000.0000 0111.0000 0000    192.168.7.0/24 
1100 0000.1010 1000.0000 1000.0000 0000    192.168.8.0/24 
1100 0000.1010 1000.0000 1001.0000 0000    192.168.9.0/24 
1100 0000.1010 1000.0000 1010.0000 0000    192.168.10.0/24 
1100 0000.1010 1000.0000 1011.0000 0000    192.168.11.0/24 
1100 0000.1010 1000.0000 1100.0000 0000    192.168.12.0/24 
1100 0000.1010 1000.0000 1101.0000 0000    192.168.13.0/24 
1100 0000.1010 1000.0000 1110.0000 0000    192.168.14.0/24 
1100 0000.1010 1000.0000 1111.0000 0000    192.168.15.0/24 
...

如果第一个网段的网络号能被n整除, 那么由它开始连续的n个网段, 能通过左移k位子网掩码进行合并

假设n=2时, k=1

yaml复制代码1100 0000.1010 1000.0000 0000.0000 0000    192.168.0.0/24       第1个网段
1100 0000.1010 1000.0000 0001.0000 0000    192.168.1.0/24       第2个网段
---------------------------------------------------------
1100 0000.1010 1000.0000 0010.0000 0000    192.168.2.0/24       第1个网段
1100 0000.1010 1000.0000 0011.0000 0000    192.168.3.0/24       第2个网段
---------------------------------------------------------
1100 0000.1010 1000.0000 0100.0000 0000    192.168.4.0/24       第1个网段
1100 0000.1010 1000.0000 0101.0000 0000    192.168.5.0/24       第2个网段
---------------------------------------------------------
1100 0000.1010 1000.0000 0110.0000 0000    192.168.6.0/24       第1个网段
1100 0000.1010 1000.0000 0111.0000 0000    192.168.7.0/24       第2个网段
---------------------------------------------------------
1100 0000.1010 1000.0000 1000.0000 0000    192.168.8.0/24       第1个网段
1100 0000.1010 1000.0000 1001.0000 0000    192.168.9.0/24       第2个网段
----------------------------------------------------------
1100 0000.1010 1000.0000 1010.0000 0000    192.168.10.0/24      第1个网段
1100 0000.1010 1000.0000 1011.0000 0000    192.168.11.0/24      第2个网段
----------------------------------------------------------
1100 0000.1010 1000.0000 1100.0000 0000    192.168.12.0/24      第1个网段
1100 0000.1010 1000.0000 1101.0000 0000    192.168.13.0/24      第2个网段
----------------------------------------------------------
1100 0000.1010 1000.0000 1110.0000 0000    192.168.14.0/24      第1个网段
1100 0000.1010 1000.0000 1111.0000 0000    192.168.15.0/24      第2个网段
...

如上面的切割方式, 每2个网段分为1组, 每组第1个网段的网络号都可以被2整除, 每组网段左移1位都可以合并成新的网段

假设n=4时, k=2

yaml复制代码1100 0000.1010 1000.0000 0000.0000 0000    192.168.0.0/24       第1个网段
1100 0000.1010 1000.0000 0001.0000 0000    192.168.1.0/24       第2个网段
1100 0000.1010 1000.0000 0010.0000 0000    192.168.2.0/24       第3个网段
1100 0000.1010 1000.0000 0011.0000 0000    192.168.3.0/24       第4个网段
---------------------------------------------------------
1100 0000.1010 1000.0000 0100.0000 0000    192.168.4.0/24       第1个网段
1100 0000.1010 1000.0000 0101.0000 0000    192.168.5.0/24       第2个网段
1100 0000.1010 1000.0000 0110.0000 0000    192.168.6.0/24       第3个网段
1100 0000.1010 1000.0000 0111.0000 0000    192.168.7.0/24       第4个网段
---------------------------------------------------------
1100 0000.1010 1000.0000 1000.0000 0000    192.168.8.0/24       第1个网段
1100 0000.1010 1000.0000 1001.0000 0000    192.168.9.0/24       第2个网段
1100 0000.1010 1000.0000 1010.0000 0000    192.168.10.0/24      第3个网段
1100 0000.1010 1000.0000 1011.0000 0000    192.168.11.0/24      第4个网段
----------------------------------------------------------
1100 0000.1010 1000.0000 1100.0000 0000    192.168.12.0/24      第1个网段
1100 0000.1010 1000.0000 1101.0000 0000    192.168.13.0/24      第2个网段
1100 0000.1010 1000.0000 1110.0000 0000    192.168.14.0/24      第3个网段
1100 0000.1010 1000.0000 1111.0000 0000    192.168.15.0/24      第4个网段
...

如上面的切割方式, 每4个网段分为1组, 每组第1个网段的网络号都可以被4整除, 每组网段左移2位都可以合并成新的网段

假设n=8时, k=3

yaml复制代码1100 0000.1010 1000.0000 0000.0000 0000    192.168.0.0/24       第1个网段
1100 0000.1010 1000.0000 0001.0000 0000    192.168.1.0/24       第2个网段
1100 0000.1010 1000.0000 0010.0000 0000    192.168.2.0/24       第3个网段
1100 0000.1010 1000.0000 0011.0000 0000    192.168.3.0/24       第4个网段
1100 0000.1010 1000.0000 0100.0000 0000    192.168.4.0/24       第5个网段
1100 0000.1010 1000.0000 0101.0000 0000    192.168.5.0/24       第6个网段
1100 0000.1010 1000.0000 0110.0000 0000    192.168.6.0/24       第7个网段
1100 0000.1010 1000.0000 0111.0000 0000    192.168.7.0/24       第8个网段
---------------------------------------------------------
1100 0000.1010 1000.0000 1000.0000 0000    192.168.8.0/24       第1个网段
1100 0000.1010 1000.0000 1001.0000 0000    192.168.9.0/24       第2个网段
1100 0000.1010 1000.0000 1010.0000 0000    192.168.10.0/24      第3个网段
1100 0000.1010 1000.0000 1011.0000 0000    192.168.11.0/24      第4个网段
1100 0000.1010 1000.0000 1100.0000 0000    192.168.12.0/24      第5个网段
1100 0000.1010 1000.0000 1101.0000 0000    192.168.13.0/24      第6个网段
1100 0000.1010 1000.0000 1110.0000 0000    192.168.14.0/24      第7个网段
1100 0000.1010 1000.0000 1111.0000 0000    192.168.15.0/24      第8个网段
...

如上面的切割方式, 每8个网段分为1组, 每组第1个网段的网络号都可以被8整除, 每组网段左移3位都可以合并成新的网段

比如:
- 第1个网段的网络以二进制0结尾, 那么它开始连续的2个网段, 能够通过左移1位子网掩码进行合并
- 第1个网段的网络以二进制00结尾, 那么它开始连续的4个网段, 能够通过左移2位子网掩码进行合并
- 第1个网段的网络以二进制000结尾, 那么它开始连续的8个网段, 能够通过左移3位子网掩码进行合并
- …

5、判断一个玩孤单是子网开始超网

首先
- 看看该网段的类型: A类网络、B类网络、C类网络?
- 默认情况下, A类子网掩码的位数是8, B类子网掩码的位数是16, C类子网掩码的位数是24
然后
- 如果该网段的子网掩码位数比默认子网掩码多, 就是子网
- 如果该网段的子网掩码位数比默认子网掩码少, 则是超网
比如
- 25.100.0.0/16 是一个A类子网
- 200.100.0.0/16 是一个C类超网

五、静态路由

在不同网段之间发消息, 需要有路由器的支持
默认情况下, 路由器只知道跟它直连的网段, 非直连的网段需要通过静态理由、动态路由告诉他
静态路由
- 管理员手动添加路由信息
- 适用于小规模网络
动态路由
- 路由器通过路由选择协议(比如: RIP、OSPF)自动获取路由信息
- 适用于于大规模网络

1、在Packet Tracer上添加设备

在Packet Tracer的桌面上添加一些设备, 如下图所示

markdown复制代码* 计算机0:
    * IP地址: 192.168.1.10
    * 子网掩码: 255.255.255.0
    * 默认网关: 192.168.1.1
* 计算机1:
    * IP地址: 192.168.2.10
    * 子网掩码: 255.255.255.0
    * 默认网关: 192.168.2.1
* 计算机2:
    * IP地址: 193.168.1.10
    * 子网掩码: 255.255.255.0
    * 默认网关: 193.168.1.1
* 计算机3:
    * IP地址: 193.168.2.10
    * 子网掩码: 255.255.255.0
    * 默认网关: 193.168.2.1
* 路由器0:
    * Fa0/0接口: 
        * IP地址: 192.168.1.1
        * 子网掩码: 255.255.255.0
    * Fa1/0接口:
        * IP地址: 192.168.2.1
        * 子网掩码: 255.255.255.0
* 路由器1:
    * Fa0/0接口: 
        * IP地址: 193.168.1.1
        * 子网掩码: 255.255.255.0
    * Fa1/0接口:
        * IP地址: 193.168.2.1
        * 子网掩码: 255.255.255.0

此时使用计算机0 ping 计算机1, 是可以连通的

但是使用计算机0 ping 计算机3, 是无法连通的

可以看到, 失败信息是由192.168.1.1返回的, 即路由器0的Fa0/0接口
此时, 我们可以通过添加静态路由的方式, 告诉路由器0: 计算机3: 193.168.2.10/24的位置

2、添加静态路由

首先设置路由器0和路由器1的Se2/0接口

此时Packet Tracer上的设备信息如下

点开路由器0, 选择静态, 就可以配置静态路由

可以看到, 有三个输入框, 分别是: 网络、掩码、下一跳
- 网络: 需要发送信息的地址网段
- 掩码: 用于计算网段
- 下一跳: 信息发送到下一个路由器的接口IP
在这里, 我们需要将计算机0发送的信息, 通过路由器0和路由器1发送给计算机3, 所以如下图所示添加

点击添加, 可以看到在网络地址下面, 生成了一条信息

同样的方式, 给路由器1也添加一条静态路由, 用于跳转到路由器0

再次使用计算机0给计算机3发送信息, 可以看到已经ping通

3、路由器的路由表

如果计算机0想要给计算机2发送消息, 只需要再给路由器0添加一条静态路由信息, 如下图所示

此时计算机0就可以给计算机1发送消息了

我们查看一下路由器0的路由表

路由器0的路由表如下图所示, 可以看到路由器每个端口配置以及路由跳转信息

4、配置静态路由中网络的几种方式

(1) 特定主机路由

网络为具体计算机的IP地址, 掩码设置为255.255.255.255

(2) 网络路由

网络为具体计算机的网段, 掩码设置为255.255.255.0

(3) 汇总路由

网络为具体计算机所在网段的超网, 掩码设置为255.255.0.0

(4) 默认路由

网络设置为0.0.0.0, 掩码设置为0.0.0.0

如下图所示

5、练习

配置静态路由, 是下面每两台计算机之间都可以通信

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Kafka 中常用的 Topic 管理操作

发表于 2021-11-23

这是我参与11月更文挑战的第23天，活动详情查看：2021最后一次更文挑战

相关：Apache Kafka 的本地部署｜一文搞定 Kafka 常见术语

如果你还不知道如何在本地部署一个 Kafka 实例作为实验环境，请参考上面的文章。

操作前记得启动 ZooKeeper 和 Kafka 服务：

1 2	bash复制代码bin/zookeeper-server-start.sh config/zookeeper.properties bin/kafka-server-start.sh config/server.properties

本文主要总结 Kafka 中用来管理 Topic 的常见操作。以下操作均在 Kafka 的目录下执行，它的脚本都在 bin/ 目录中（针对 Windows 操作系统的 .bat 文件在 bin/windows/ 中）。

针对 Topic 进行操作的指令，大多是通过 bin/kafka-topics.sh 来进行，执行这个脚本时，需要通过 --bootstrap-server 指定要连接到的 Kafka Broker，其后是具体的操作。

创建主题

创建主题的操作如下：

1	ruby复制代码bin/kafka-topics.sh --bootstrap-server <host>:<port> --create --topic <topic_name> --partitions 1 --replication-factor 1

其中：

--bootstrap-server <host>:<port> 指定要连接的 Kafka Broker，这里需要提供 Kafka Broker 的服务地址和端口号。
--create 代表要执行「创建」操作。
--topic <topic_name> 指定即将创建的主题的名称
--partitions 1 指定主题的分区数。
--replication-factor 1 指定副本数。

查询主题

查询已经创建好的主题列表：

1	ruby复制代码bin/kafka-topics.sh --bootstrap-server <host>:<port> --list

把创建操作的 --create 及之后的内容，替换成 --list 即可。

如果想要查看某个主题的详细信息，则可以执行：

1	ruby复制代码bin/kafka-topics.sh --bootstrap-server <host>:<port> --describe --topic <topic_name>

执行结果类似一下内容：

1 2	yaml复制代码Topic: fooTopic TopicId: EXpCrwSuTDCiuBtLh2V9gg PartitionCount: 1 ReplicationFactor: 1 Configs: segment.bytes=1073741824 Topic: fooTopic Partition: 0 Leader: 0 Replicas: 0 Isr: 0

包含了主题的详细信息以及各个分区的信息，这个命令需要指定查询的主题名称，如果不提供的话，则会展示所有可见的主题的详细信息。

增加主题的分区数

修改主题分区数的操作如下：

1	xml复制代码bin/kafka-topics.sh --bootstrap-server <host>:<port> --alter --topic <topic_name> --partitions <新分区数>

执行命令后，如果控制台什么都没有出现，则证明执行成功了，可以使用 --describe 来查看一下结果：

1
2
3

yaml复制代码Topic: fooTopic	TopicId: EXpCrwSuTDCiuBtLh2V9gg	PartitionCount: 2	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: fooTopic	Partition: 0	Leader: 0	Replicas: 0	Isr: 0
	Topic: fooTopic	Partition: 1	Leader: 0	Replicas: 0	Isr: 0

之所以把这个操作叫做「增加主题的分区数」而不是「修改主题的分区数」，是因为这里指定的分区数，一定要比现有的分区数大，否则会报错。报错信息类似一下内容：

1 2	vbnet复制代码ERROR org.apache.kafka.common.errors.InvalidPartitionsException: Topic currently has 2 partitions, which is higher than the requested 1. (kafka.admin.TopicCommand$)

删除主题

删除主题使用 --delete 操作，需要指定要删除的主题名称；

1	ruby复制代码bin/kafka-topics.sh --bootstrap-server <host>:<port> --delete --topic <topic_name>

这里值得注意的是，如果有副本所在的 Broker 处于宕机的状态，是无法完成删除的。

修改 Topic 级别的参数

1	xml复制代码bin/kafka-configs.sh --zookeeper <host>:<port> --entity-type topics --entity-name <topic_name> --alter --add-config <参数名>=<参数值>

注意这里并没有通过 --bootstrap-server 指定 Broker 节点的连接，而至通过 --zookeeper 指定了 ZooKeeper 的连接。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

从头到尾，通透I/O模型内核态和用户态同步阻塞 I/O

发表于 2021-11-23

你好，我是yes。

上篇我们已经搞懂了 socket 的通信内幕，也明白了网络 I/O 确实会有很多阻塞点，阻塞 I/O 随着用户数的增长只能利用增加线程的方式来处理更多的请求，而线程不仅会占用内存资源且太多的线程竞争会导致频繁地上下文切换产生巨大的开销。

因此，阻塞 I/O 已经不能满足需求，所以后面大佬们不断地优化和演进，提出了多种 I/O 模型。

在 UNIX 系统下，一共有五种 I/O 模型，今天我们就来盘一盘它！

不过在介绍 I/O 模型之前，我们需要先了解一下前置知识。

内核态和用户态

我们的电脑可能同时运行着非常多的程序，这些程序分别来自不同公司。

谁也不知道在电脑上跑着的某个程序会不会发疯似得做一些奇怪的操作，比如定时把内存清空了。

因此 CPU 划分了非特权指令和特权指令，做了权限控制，一些危险的指令不会开放给普通程序，只会开放给操作系统等特权程序。

你可以理解为我们的代码调用不了那些可能会产生“危险”操作，而操作系统的内核代码可以调用。

这些“危险”的操作指：内存的分配回收，磁盘文件读写，网络数据读写等等。

如果我们想要执行这些操作，只能调用操作系统开放出来的 API ，也称为系统调用。

这就好比我们去行政大厅办事，那些敏感的操作都由官方人员帮我们处理（系统调用），所以道理都是一样的，目的都是为了防止我们(普通程序)乱来。

这里又有两个名词：

用户空间
内核空间。

我们普通程序的代码是跑在用户空间上的，而操作系统的代码跑在内核空间上，用户空间无法直接访问内核空间的。当一个进程运行在用户空间时就处于用户态，运行在内核空间时就处于内核态。

当处于用户空间的程序进行系统调用，也就是调用操作系统内核提供的 API 时，就会进行上下文的切换，切换到内核态中，也时常称之为陷入内核态。

那为什么开头要先介绍这个知识点呢？

因为当程序请求获取网络数据的时候，需要经历两次拷贝：

程序需要等待数据从网卡拷贝到内核空间。
因为用户程序无法访问内核空间，所以内核又得把数据拷贝到用户空间，这样处于用户空间的程序才能访问这个数据。

介绍这么多就是让你理解为什么会有两次拷贝，且系统调用是有开销的，因此最好不要频繁调用。

然后我们今天说的 I/O 模型之间的差距就是这拷贝的实现有所不同！

今天我们就以 read 调用，即读取网络数据为例子来展开 I/O 模型。

发车！

同步阻塞 I/O

当用户程序的线程调用 read 获取网络数据的时候，首先这个数据得有，也就是网卡得先收到客户端的数据，然后这个数据有了之后需要拷贝到内核中，然后再被拷贝到用户空间内，这整一个过程用户线程都是被阻塞的。

假设没有客户端发数据过来，那么这个用户线程就会一直阻塞等着，直到有数据。即使有数据，那么两次拷贝的过程也得阻塞等着。

所以这称为同步阻塞 I/O 模型。

它的优点很明显，简单。调用 read 之后就不管了，直到数据来了且准备好了进行处理即可。

缺点也很明显，一个线程对应一个连接，一直被霸占着，即使网卡没有数据到来，也同步阻塞等着。

我们都知道线程是属于比较重资源，这就有点浪费了。

所以我们不想让它这样傻等着。

于是就有了同步非阻塞 I/O。

同步非阻塞 I/O

从图中我们可以很清晰的看到，同步非阻塞I/O 基于同步阻塞I/O 进行了优化：

在没数据的时候可以不再傻傻地阻塞等着，而是直接返回错误，告知暂无准备就绪的数据！

这里要注意，从内核拷贝到用户空间这一步，用户线程还是会被阻塞的。

这个模型相比于同步阻塞 I/O 而言比较灵活，比如调用 read 如果暂无数据，则线程可以先去干干别的事情，然后再来继续调用 read 看看有没有数据。

但是如果你的线程就是取数据然后处理数据，不干别的逻辑，那这个模型又有点问题了。

等于你不断地进行系统调用，如果你的服务器需要处理海量的连接，那么就需要有海量的线程不断调用，上下文切换频繁，CPU 也会忙死，做无用功而忙死。

那怎么办？

于是就有了I/O 多路复用。

I/O 多路复用

从图上来看，好像和上面的同步非阻塞 I/O 差不多啊，其实不太一样，线程模型不一样。

既然同步非阻塞 I/O 在太多的连接下频繁调用太浪费了，那就招个专员吧。

这个专员工作就是管理多个连接，帮忙查看连接上是否有数据已准备就绪。

也就是说，可以只用一个线程查看多个连接是否有数据已准备就绪。

具体到代码上，这个专员就是 select ，我们可以往 select 注册需要被监听的连接，由 select 来监控它所管理的连接是否有数据已就绪，如果有则可以通知别的线程来 read 读取数据，这个 read 和之前的一样，还是会阻塞用户线程。

这样一来就可以用少量的线程去监控多条连接，减少了线程的数量，降低了内存的消耗且减少了上下文切换的次数，很舒服。

想必到此你已经理解了什么叫 I/O 多路复用。

所谓的多路指的是多条连接，复用指的是用一个线程就可以监控这么多条连接。

看到这，你再想想，还有什么地方可以优化的？

信号驱动式I/O

上面的 select 虽然不阻塞了，但是他得时刻去查询看看是否有数据已经准备就绪，那是不是可以让内核告诉我们数据到了而不是我们去轮询呢？

信号驱动 I/O 就能实现这个功能，由内核告知数据已准备就绪，然后用户线程再去 read（还是会阻塞）。

听起来是不是比 I/O 多路复用好呀？那为什么好像很少听到信号驱动 I/O？

为什么市面上用的都是 I/O 多路复用而不是信号驱动?

因为我们的应用通常用的都是 TCP 协议，而 TCP 协议的 socket 可以产生信号事件有七种。

也就是说不仅仅只有数据准备就绪才会发信号，其他事件也会发信号，而这个信号又是同一个信号，所以我们的应用程序无从区分到底是什么事件产生的这个信号。

那就麻了呀！

所以我们的应用基本上用不了信号驱动 I/O，但如果你的应用程序用的是 UDP 协议，那是可以的，因为 UDP 没这么多事件。

因此，这么一看对我们而言信号驱动 I/O 也不太行。

异步 I/O

信号驱动 I/O 虽然对 TCP 不太友好，但是这个思路对的：往异步发展，但是它并没有完全异步，因为其后面那段 read 还是会阻塞用户线程，所以它算是半异步。

因此，我们得想下如何弄成全异步的，也就是把 read 那步阻塞也省了。

其实思路很清晰：让内核直接把数据拷贝到用户空间之后再告知用户线程，来实现真正的非阻塞I/O！

所以异步 I/O 其实就是用户线程调用 aio_read ，然后包括将数据从内核拷贝到用户空间那步，所有操作都由内核完成，当内核操作完毕之后，再调用之前设置的回调，此时用户线程就拿着已经拷贝到用户控件的数据可以继续执行后续操作。

在整个过程中，用户线程没有任何阻塞点，这才是真正的非阻塞I/O。

那么问题又来了:

为什么常用的还是I/O多路复用，而不是异步I/O？

因为 Linux 对异步 I/O 的支持不足，你可以认为还未完全实现，所以用不了异步 I/O。

这里可能有人会说不对呀，像 Tomcat 都实现了 AIO的实现类，其实像这些组件或者你使用的一些类库看起来支持了 AIO(异步I/O)，实际上底层实现是用 epoll 模拟实现的。

而 Windows 是实现了真正的 AIO，不过我们的服务器一般都是部署在 Linux 上的，所以主流还是 I/O 多路复用。

最后

至此，想必你已经清晰五种 I/O 模型是如何演进的了。

下篇，我将讲讲谈到网络 I/O 经常会伴随的几个容易令人混淆的概念：同步、异步、阻塞、非阻塞。

参考：

我是yes，从一点点到亿点点，欢迎关注我的个人公众号【yes的练级攻略】，我们下篇见~

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Go语言学习查缺补漏ing Day5

发表于 2021-11-23

「这是我参与11月更文挑战的第23天，活动详情查看：2021最后一次更文挑战」。

Go语言学习查缺补漏ing Day5

本文收录于我的专栏：《让我们一起Golang》

一、将切片当可变参数传递的一个问题

我们来看一段代码：

go复制代码package main

import "fmt"

func Myfunc(num ...int) {
    num[0] = 2
    total := 0
    for _, i := range num {
        total += i
    }
    num[1] = total
}
func Sum(num ...int) {

    total := 0
    for _, i := range num {
        total += i
    }
    num[1] = total
}
func main() {
    i := []int{1, 2, 3}
    Myfunc(i...)
    fmt.Println(i)
    Sum(i...)
    fmt.Println(i)
}

这段代码的运行结果是：

1 2	csharp复制代码[2 7 3] [2 12 3]

是不是很奇怪为什么切片内的值改变了？其实我们将切片作为函数参数传递给函数是进行的值传递，所以我们传递给函数的参数其实是下面这个slice的值拷贝。

go复制代码type slice struct{
    value *int 
    length uint 
    capacity uint 
}

但是，这个slice结构体内的value是一个指针，所以这个slice就算进行了拷贝，它和它的拷贝值都是同一个value，指向同一块区域。

所以上面的代码，我们可以在Myfunc和Sum函数中对main函数内的i切片进行操作。

但是，当我们对num切片进行扩容操作时，拷贝值的value指向的地址就可能会发生变化。

比如我们进行下面的操作：

go复制代码func Sum(num ...int) {
    num[0] = 8
    total := 0
    for _, i := range num {
        total += i
    }
    num = append(num, total)
    num[1] = total
}

运行结果是：

1 2	go复制代码[2 7 3] [8 7 3]

这就侧面描述了slice的扩容算法。

另外说到可变参数，我们还需要注意几点，这里我们应该了解可变长参数应该是函数头中最后一个参数！！！

有些人会疑惑，不是说可变参数吗？怎么传入了一个切片？其实可变参数的底层就是用切片实现的，它是将传入的一个或多个参数转换为一个切片。

二、Go中不允许不同类型的数据进行运算

我们都知道，Go中具有极其严格的静态类型限制，只有相同类型的数据才能进行运算，那么如何解决这个问题呢？

第一点自然是强制类型转换。比如:

go复制代码func main() {
    var (
        i int     = 3
        j float32 = 3.1
    )
    fmt.Println(float32(i) + j)
}

这个程序运行就不会编译错误，运行结果是6.1。

还有一种方法，不过适用场景比较单一。就是定义无类型常量。比如：

go复制代码package main

import (
    "fmt"
)

func main() {
    const i = 1
    const j = 2.1

    fmt.Println(i + j)
}

运行就不会报错，运行结果是3.1.

三、不同长度数组能否进行比较

看一看下面这段代码：

go复制代码package main

import (
    "fmt"
)

func main() {
    a := [2]int{1, 2}
    b := [3]int{1, 2}
    if a == b {
        fmt.Println("equal")
    } else {
        fmt.Println("not equal")
    }
}

你觉得他们相不相等？

哈哈哈哈

其实都不能通过编译，因为数组的长度是数组的一部分，所以两个数组不是相同的类型，所以他们无法进行比较，故会报错:

1	go复制代码invalid operation: a == b (mismatched types [2]int and [3]int)

四、关于map的一点需要注意的地方

我们先来看看这段代码：

go复制代码package main

import (
    "fmt"
)

func main() {
    Map := make(map[string]int)
    delete(Map, "map")
    fmt.Println(Map["map"])
}

Map是空的，你觉得会报错吗？如果不报错，会输出什么？

答案是不会报错，会输出0.

是不是想对了？下面来解释一下：

Go设计的就是删除map中不存在的键值对时不会报错，另外获取map中并不存在的键值对时，获取到的是值类型的零值。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

关于 RocketMQ ClientID 相同引发的消息堆积

发表于 2021-11-23

首先，造成这个问题的 BUG RocketMQ 官方已经在 3月16号 的这个提交中修复了，这里只是探讨一下在修复之前造成问题的具体细节，更多的上下文可以参考我之前写的《RocketMQ Consumer 启动时都干了些啥？》，这篇文章讲解了 RocketMQ 的 Consumer 启动之后都做了哪些操作，对理解本次要讲解的 BUG 有一定的帮助。

其中讲到了：

消息堆积

重复消费自不必说，你 ClientID 都相同了。本篇着重聊聊为什么会消息堆积。

文章中讲到，初始化 Consumer 时，会初始化 Rebalance 的策略。你可以大致将 Rebalance 策略理解为如何将一个 Topic 下的 m 个 MessageQueue 分配给一个 ConsumerGroup 下的 n 个 Consumer 实例的策略，看着有些绕，其实就长这样：

rebalance策略

而从 Consumer 初始化的源码中可以看出，默认情况下 Consumer 采取的 Rebalance 策略是 AllocateMessageQueueAverage()。

默认的 Rebalance 策略

默认的策略很好理解，将 MessageQueue 平均的分配给 Consumer。举个例子，假设有 8 个 MessageQueue，2 个 Consumer，那么每个 Consumer 就会被分配到 4 个 MessageQueue。

那如果分配不均匀怎么办？例如只有 7 个 MessageQueue，但是 Consumer 仍然是 2 个。此时 RocketMQ 会将多出来的部分，对已经排好序的 Consumer 再做平均分配，一个一个分发给 Consumer，直到分发完。例如刚刚说的 7 个 MessageQueue 和 2 个 ConsumerGroup 这种 case，排在第一个的 Consumer 就会被分配到 4 个 MessageQueue，而第二个会被分配到 3 个 MessageQueue。

大家可以先理解一下 AllocateMessageQueueAveragely 的实现，作为默认的 Rebalance 的策略，其实现位于这里：

默认策略的实现位置

接下来我们看看，AllocateMessageQueueAveragely 内部具体都做了哪些事情。

其核心其实就是实现的 AllocateMessageQueueStrategy 接口中的 allocate 方法。实际上，RocketMQ 对该接口总共有 5 种实现：

AllocateMachineRoomNearby
AllocateMessageQueueAveragely
AllocateMessageQueueAveragelyByCircle
AllocateMessageQueueByConfig
AllocateMessageQueueByMachineRoom
AllocateMessageQueueConsistentHash

其默认的 AllocateMessageQueueAveragely 只是其中的一种实现而已，那执行 allocate 它需要什么参数呢？

需要以下四个：

ConsumerGroup 消费者组的名字
currentCID 当前消费者的 clientID
mqAll 当前 ConsumerGroup 所消费的 Topic 下的所有的 MessageQueue
cidAll 当前 ConsumerGroup 下所有消费者的 ClientID

实际上是将某个 Topic 下的所有 MessageQueue 分配给属于同一个消费者的所有消费者实例，粒度是 By Topic 的。

所以到这里剩下的事情就很简单了，无非就是怎么样把这一堆 MessageQueue 分配给这一堆 Consumer。这个怎么样，就对应了 AllocateMessageQueueStrategy 的不同实现。

接下来我们就来看看 AllocateMessageQueueAveragely 是如何对 MessageQueue 进行分配的，之前讲源码我一般都会一步一步的来，结合源码跟图，但是这个源码太短了，我就直接先给出来吧。

java复制代码public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
  if (currentCID == null || currentCID.length() < 1) {
    throw new IllegalArgumentException("currentCID is empty");
  }
  if (mqAll == null || mqAll.isEmpty()) {
    throw new IllegalArgumentException("mqAll is null or mqAll empty");
  }
  if (cidAll == null || cidAll.isEmpty()) {
    throw new IllegalArgumentException("cidAll is null or cidAll empty");
  }

  List<MessageQueue> result = new ArrayList<MessageQueue>();

  // 判断一下当前的客户端是否在 cidAll 的集合当中
  if (!cidAll.contains(currentCID)) {
    log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
             consumerGroup,
             currentCID,
             cidAll);
    return result;
  }

  // 拿到当前消费者在所有的消费者实例数组中的位置
  int index = cidAll.indexOf(currentCID);
  // 用 messageQueue 的数量 对 消费者实例的数量取余数, 这个实际上就把不够均匀分的 MessageQueue 的数量算出来了
  // 举个例子, 12 个 MessageQueue, 有 5 个 Consumer, 12 % 5 = 2 
  int mod = mqAll.size() % cidAll.size();
  int averageSize =
    mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size());
  int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
  int range = Math.min(averageSize, mqAll.size() - startIndex);
  for (int i = 0; i < range; i++) {
    result.add(mqAll.get((startIndex + i) % mqAll.size()));
  }
  return result;
}

其实前半部分都是些常规的 check，可以忽略不看，从这里：

1	java复制代码int index = cidAll.indexOf(currentCID);

开始，才是核心逻辑。为了避免逻辑混乱，还是假设有 12 个 MessageQueue，5 个 Consumer，同时假设 index=0 。

那么 mod 的值就为 12 % 5 = 2 了。

而 averageSize 的值，稍微有点绕。如果 MessageQueue 的数量比消费者的数量还少，那么就为 1 ；否则，就走这一堆逻辑(mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size())。我们 index 是 0，而 mod 是 2，index < mod 则是成立的，那么最终 averageSize 的值就为 12 / 5 + 1 = 3。

接下来是 startIndex，由于这个三元运算符的条件是成立的，所以其值为 0 * 3 ，就为 0。

看了一大堆逻辑，是不是已经晕了？直接举实例：

12 个 Message Queue

5 个 Consumer 实例

按照上面的分法：

排在第 1 的消费者分到 3 个

排在第 2 的消费者分到 3 个

排在第 3 的消费者分到 2 个

排在第 4 的消费者分到 2 个

排在第 5 的消费者分到 2 个

具体分配流程

所以，你可以大致认为：

先“均分”，12 / 5 取整为 2。然后“均分”完之后还剩下 2 个，那么就从上往下，挨个再分配，这样第 1、第 2 个消费者就会被多分到 1 个。

所以如果有 13 个 MessageQueue，5 个 Consumer，那么第 1、第 2、第 3 就会被分配 3 个。

但并不准确，因为分配的 MessageQueue 是一次性的，例如那 3 个 MessageQueue 是一次性获取的，不会先给 2 个，再给 1 个。

而我们开篇提到的 Consumer 的 ClientID 相同，会造成什么？

当然是 index 的值相同，进而造成 mod、averageSize、startIndex、range 全部相同。那么最后 result.add(mqAll.get((startIndex + i) % mqAll.size())); 时，本来不同的 Consumer，会取到相同的 MessageQueue（举个例子，Consumer 1 和 Consumer 2 都取到了前 3 个 MessageQueue），从而造成有些 MessageQueue（如果有的话）没有 Consumer 对其消费，而没有被消费，消息也在不停的投递进来，就会造成消息的大量堆积。

当然，现在的新版本从代码上看已经修复这个问题了，这个只是对之前的版本的原因做一个探索。

本篇文章已放到我的 Github github.com/sh-blog 中，欢迎 Star。微信搜索关注【SH的全栈笔记】

如果你觉得这篇文章对你有帮助，还麻烦点个赞，关个注，分个享，留个言。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有