文章目录
写在前面的话
ROS 的通信机制
话题通信机制
服务通信机制
共享参数机制
话题与服务的区别

写在前面的话


文档没有任何商业因素，本着共享的精神进行分享，如有素材侵权，请给我留言;
文档都是自己平时看书或工作中的笔记，观点错误的地方欢迎留言；



ROS 的通信机制

ROS 最常用的通信方式有三种：话题通信、服务通信 和 共享参数，当然还有其他的通信方式，例如，节点之间的 Nodelet 通信（这种通信方式主要是为了保证数据实时性）；


话题通信机制

在话题（Topic）通信机制里，主要要三个角色：发布者，订阅者，Master；根据 Master 的参与主要分为两个阶段：连接前的准备，连接和通信。下图为单个发布者 与 订阅者 的通讯示意图；




第一阶段：建立前的准备

在建立连接阶段，主要是解决发布者 与 订阅者 进行匹配进而连接的问题；在这个阶段主要分为五步：

发布者（Talker）启动，通过RPC向 ROS Master 注册发布者的信息，包括：发布者节点信息，话题名，话题缓存大小等；Master 会将这些信息加入注册列表中；
订阅者（Listener）启动，通过 RPC 向 ROC Master 注册订阅者信息，包括：订阅者节点信息，话题名等；Master 会将这些信息加入注册列表；
Master 进行节点匹配：Master 会根据订阅者提供的信息，在注册列表中查找匹配的发布者；如果没有发布者（Talker），则等待发布者（Talker）的加入；如果找到匹配的发布者（Talker），则会主动把发布者（Talker）（有可能是很多个 Talker）的地址通过 RPC 传送给订阅者（Listener）节点；
Listener 接收到 Master 的发出的 Talker 的地址信息，尝试通过 RPC 向 Talker 发出连接请求（信息包括：话题名，消息类型以及通讯协议（TCP/UDP））；
Talker 收到 Listener 发出的连接请求后，通过 RPC 向 Listener 确认连接请求（包含的信息为自身 TCP 地址信息）；

至此，Talker 和 Listener 做好了连接前的准备工作。在这个过程中，Master 起到了 牵线搭桥 的作用；



第二阶段：连接和通信

在发布消息阶段，主要解决的是发布者(Talker) 如何 发布消息进入传递给 订阅者(Listener) 的过程。在这个过程中，完全是 Talker 和 Listener 两者之间的信息单向流动，Master 并未参与其中；主要分为两步：

Listener 接收到 Talker 的确认消息后，使用 TCP 尝试与 Talker 建立网络连接；
成功连接之后，Talker 开始向 Listener 发布话题消息数据；

至此，完成了 Talker 向 Listener 发布消息的过程；Master 在这个阶段并不参与两者之间的数据传递；

需要注意的是，有可能多个 Talker 连接一个 Listener，也有可能是一个 Talker 连接上多个 Listener；


服务通信机制

服务是基于 C/S 模式的双向数据传输模式（有应答的通信机制），话题是无应答的通信机制。根据 Master 是否参数也分为两部分：连接前的准备，连接和通信；通信流程如下：






连接前的准备


发布者（Talker）启动，通过RPC向 ROS Master 注册发布者的信息，包括：发布者节点信息，话题名，话题缓存大小等；Master 会将这些信息加入注册列表中；
订阅者（Listener）启动，通过 RPC 向 ROC Master 注册订阅者信息，包括：订阅者节点信息，话题名等；Master 会将这些信息加入注册列表；
Master 进行节点匹配：Master 会根据订阅者提供的信息，在注册列表中查找匹配的发布者；如果没有发布者（Talker），则等待发布者（Talker）的加入；如果找到匹配的发布者（Talker），则会主动把发布者（Talker）（有可能是很多个 Talker）的地址通过 RPC 传送给订阅者（Listener）节点；

至此，连接前的准备工作就做完了，可以发现服务通信的 连接前的准备 工作比话题通信的 连接前的准备 工作少两步；



连接和通信

这一部分和话题通信的 连接和通信 步骤都是一样的。唯一不同的是服务通信是有应答的。

Listener 接收到 Talker 的确认消息后，使用 TCP 尝试与 Talker 建立网络连接；
成功连接之后，Talker 开始向 Listener 发布话题消息数据；

至此，服务通信完成连接和通信过程，可以进行有应答的信息交流了。

需要注意的是，有可能是一个 Talker 连接上多个 Listener；


共享参数机制

参数共享机制类似于程序中的全局变量，Talker 去更新全局变量（共享的参数），Listener 去获取更新后的全局变量（共享的参数）；这个通信过程不涉及 TCP/UDP 的通信；示意图如下：




Talker 更新全局变量；Talker 通过 RPC 更新 ROS Master 中的共享参数（包含参数名和参数值）；
Listener 通过 RPC 向 ROS Master 发送参数查询请求（包含要查询的参数名）；
ROS Master 通过 RPC 回复 Listener 的请求（包括参数值）；

从上述流程可知，如果 Listener 向实时知道共享参数的变化，需要自己不停的去询问 ROS Master；


话题与服务的区别

话题和服务是 ROS 中使用最多的通信方法，它们之间有很多不同之处：





条目
话题
服务




同步性
异步
同步


通信模型
发布/订阅
客户端/服务器


反馈机制
无
有


底层协议
ROSTCP/ROSUDP
ROSTCP/ROSUDP


缓冲区
有
无


实时性
弱
强


节点关系
多对多
一对多（一个Server）


使用场景
弱逻辑处理，多数据传输
强逻辑处理，少数据传输



话题是 ROS 中基于发布者/订阅者模型的异步通信，发布者与订阅者双方解耦，常用于不断更新，含有较少逻辑处理的数据通信；

服务是 ROS 中基于客户端。服务器模型的同步通信，适用于逻辑性强的数据交换；

Worker间的通信经常需要通过网络跨节点进行，Storm使用ZeroMQ或Netty(0.9以后默认使用)作为进程间通信的消息框架。 

Worker进程内部通信：不同worker的thread通信使用LMAX Disruptor来完成。 

    不同topologey之间的通信，Storm不负责，需要自己想办法实现，例如使用kafka等；

1、Worker进程间通信

worker进程间消息传递机制，消息的接收和处理的大概流程见下图 


   对于worker进程来说，为了管理流入和传出的消息，每个worker进程有一个独立的接收线程(对配置的TCP端口supervisor.slots.ports进行监听); 

对应Worker接收线程，每个worker存在一个独立的发送线程，它负责从worker的transfer-queue中读取消息，并通过网络发送给其他worker 

   每个executor有自己的incoming-queue和outgoing-queue。 

Worker接收线程将收到的消息通过task编号传递给对应的executor(一个或多个)的incoming-queues; 

每个executor有单独的线程分别来处理spout/bolt的业务逻辑，业务逻辑输出的中间数据会存放在outgoing-queue中，当executor的outgoing-queue中的tuple达到一定的阀值，executor的发送线程将批量获取outgoing-queue中的tuple,并发送到transfer-queue中。 

  每个worker进程控制一个或多个executor线程，用户可在代码中进行配置。其实就是我们在代码中设置的并发度个数。

2、Worker进程间通信分析



1、  Worker接受线程通过网络接受数据，并根据Tuple中包含的taskId，匹配到对应的executor；然后根据executor找到对应的incoming-queue，将数据存发送到incoming-queue队列中。 

2、  业务逻辑执行现成消费incoming-queue的数据，通过调用Bolt的execute(xxxx)方法，将Tuple作为参数传输给用户自定义的方法 

3、  业务逻辑执行完毕之后，将计算的中间数据发送给outgoing-queue队列，当outgoing-queue中的tuple达到一定的阀值，executor的发送线程将批量获取outgoing-queue中的tuple,并发送到Worker的transfer-queue中 

4、  Worker发送线程消费transfer-queue中数据，计算Tuple的目的地，连接不同的node+port将数据通过网络传输的方式传送给另一个的Worker。 

5、  另一个worker执行以上步骤1的操作。

3、Worker进程间技术(Netty、ZeroMQ)

3.1、Netty

Netty是一个NIO client-server(客户端服务器)框架，使用Netty可以快速开发网络应用，例如服务器和客户端协议。Netty提供了一种新的方式来使开发网络应用程序，这种新的方式使得它很容易使用和有很强的扩展性。Netty的内部实现时很复杂的，但是Netty提供了简单易用的api从网络处理代码中解耦业务逻辑。Netty是完全基于NIO实现的，所以整个Netty都是异步的。 

书籍：Netty权威指南

3.2、ZeroMQ

ZeroMQ是一种基于消息队列的多线程网络库，其对套接字类型、连接处理、帧、甚至路由的底层细节进行抽象，提供跨越多种传输协议的套接字。ZeroMQ是网络通信中新的一层，介于应用层和传输层之间（按照TCP/IP划分），其是一个可伸缩层，可并行运行，分散在分布式系统间。 

ZeroMQ定位为：一个简单好用的传输层，像框架一样的一个socket library，他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ的明确目标是“成为标准网络协议栈的一部分，之后进入Linux内核”。

4、Worker 内部通信技术(Disruptor)

4.1、   Disruptor的来历

一个公司的业务与技术的关系，一般可以分为三个阶段。第一个阶段就是跟着业务跑。第二个阶段是经历了几年的时间，才达到的驱动业务阶段。第三个阶段，技术引领业务的发展乃至企业的发展。所以我们在学习Disruptor这个技术时，不得不提LMAX这个机构，因为Disruptor这门技术就是由LMAX公司开发并开源的。 

LMAX是在英国注册并受到FSA监管（监管号码为509778）的外汇黄金交易所。LMAX也是欧洲第一家也是唯一一家采用多边交易设施Multilateral Trading Facility（MTF）拥有交易所牌照和经纪商牌照的欧洲顶级金融公司 

 LAMX拥有最迅捷的交易平台，顶级技术支持。LMAX交易所使用“（MTF）分裂器Disruptor”技术，可以在极短时间内（一般在3百万秒之一内）处理订单，在一个线程里每秒处理6百万订单。所有订单均为撮合成交形式，无一例外。多边交易设施（MTF）曾经用来设计伦敦证券交易 所（london Stock Exchange）、德国证券及衍生工具交易所（Deutsche Borse）和欧洲证券交易所（Euronext）。 

 2011年LMAX凭借该技术获得了金融行业技术评选大赛的最佳交易系统奖和甲骨文“公爵杯”创新编程框架奖。

4.2、Disruptor是什么

1、  简单理解：Disruptor是一个Queue。Disruptor是实现了“队列”的功能，而且是一个有界队列。而队列的应用场景自然就是“生产者-消费者”模型。 

2、  在JDK中Queue有很多实现类，包括不限于ArrayBlockingQueue、LinkBlockingQueue，这两个底层的数据结构分别是数组和链表。数组查询快，链表增删快，能够适应大多数应用场景。 

3、  但是ArrayBlockingQueue、LinkBlockingQueue都是线程安全的。涉及到线程安全，就会有synchronized、lock等关键字，这就意味着CPU会打架。  

4、  Disruptor一种线程之间信息无锁的交换方式（使用CAS（Compare And Swap/Set）操作）。

4.3、Disruptor主要特点

1、  没有竞争=没有锁=非常快。 

2、  所有访问者都记录自己的序号的实现方式，允许多个生产者与多个消费者共享相同的数据结构。 

3、  在每个对象中都能跟踪序列号（ring buffer，claim Strategy，生产者和消费者），加上神奇的cache line padding，就意味着没有为伪共享和非预期的竞争。

4.4、   Disruptor 核心技术点

Disruptor可以看成一个事件监听或消息机制，在队列中一边生产者放入消息，另外一边消费者并行取出处理. 

底层是单个数据结构：一个ring buffer。 

每个生产者和消费者都有一个次序计算器，以显示当前缓冲工作方式。 

每个生产者消费者能够操作自己的次序计数器的能够读取对方的计数器，生产者能够读取消费者的计算器确保其在没有锁的情况下是可写的。

核心组件 

   Ring Buffer 环形的缓冲区，负责对通过 Disruptor 进行交换的数据（事件）进行存储和更新。 

   Sequence 通过顺序递增的序号来编号管理通过其进行交换的数据（事件），对数据(事件)的处理过程总是沿着序号逐个递增处理。 

   RingBuffer底层是个数组，次序计算器是一个64bit long 整数型，平滑增长。 


1、  接受数据并写入到脚标31的位置，之后会沿着序号一直写入，但是不会绕过消费者所在的脚标。 

2、  Joumaler和replicator同时读到24的位置，他们可以批量读取数据到30 

3、消费逻辑线程读到了14的位置，但是没法继续读下去，因为他的sequence暂停在15的位置上，需要等到他的sequence给他序号。如果sequence能正常工作，就能读取到30的数据。

转自：http://blog.csdn.net/tototuzuoquan/article/details/73551560

一  java实现线程通信机制主要通过以下几种方式

1 while()轮询

2 等待/通知机制

wait()、notify()、notifyAll()结合synchronized关键字

3 条件对象：Condition

Condition和ReentrantLock重入锁

 

 

此外,java为线程间通信还提供了一些关键字：synchronized、volatile、final,这点我们最后在做介绍

 

第一种方式：while()轮询

第二种方式：等待/通知机制

synchronized是JVM提供的内置锁,我们通过java中的生产者消费者模式来看看wait/notify是如何实现线程间通信的

wait()/notify()/notifyAll()是Object类定义的方法,必须与synchronized一起使用,更确切的来说是在synchronized临界区内使用,

也就是说调用wait()方法的前提是该线程已经获取锁,调用notify()/notifyAll()的前提是也是该线程已经获取了锁,

wait() : 使当前线程进入阻塞状态,退出临界区,释放CPU,暂时放弃所获取的锁,当其他线程调用notify()/notidyAll()方法的时候,会唤醒一个或者多个处于等待该线程可能会被唤醒然后重新取竞争获取锁,如果重新获得了锁,被唤醒的线程会继续从上次阻塞的地方执行。

notify()/notifyAll():当前线程调用此方法,会唤醒之前阻塞的线程,但是notify()/notifyAll()方法不会释放锁,而是会继续往下执行,直到遇到wait()方法才会释放锁,所以一般线程调用完成notify()/notifyAll()方法后,都会调用wait()释放锁,退出临界区,从而让其他线程有机会获取锁

package com.demo.lock1;

import java.util.ArrayList;
import java.util.List;

public class ProducerConsumer {
	public static void main(String[] args) {
		List<Integer> list = new ArrayList<Integer>();
		Producer producer = new Producer(list);
		Consumer consumer = new Consumer(list);
		producer.start();
		consumer.start();

	}
}

/**
 * 生产者
 * 
 * @author jiangry01
 *
 */
class Producer extends Thread {

	private List<Integer> list;// 仓库

	public Producer(List<Integer> list) {
		this.list = list;
	}

	public void run() {
		while (true) {
			synchronized (list) {
				while (list.isEmpty()) {
					for (int i = 0; i < 10; i++) {
						list.add(i);
					}
					System.out.println("线程-" + Thread.currentThread().getName() + "-生产-" + list.size() + " :条消息");
				}
				list.notify();
				//list.clear();
//				try {
//					list.wait();
//				} catch (InterruptedException e) {
//					e.printStackTrace();
//				}
			}
		}
	}
}

/**
 * 消费者
 * 
 * @author jiangry01
 *
 */
class Consumer extends Thread {

	private List<Integer> list;// 仓库

	public Consumer(List<Integer> list) {
		this.list = list;
	}

	public void run() {
		while (true) {
			synchronized (list) {
				while (!list.isEmpty()) {
					int size = list.size();
					for (int i = size-1; i >= 0; i--) {
						System.out.println(list.remove(i));
						try {
							Thread.sleep(1000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					System.out.println("线程-" + Thread.currentThread().getName() + "-仓库数据-" + list.size() + " :条消息");
				}
				list.notify();
//				try {
//					list.wait();
//				} catch (InterruptedException e) {
//					e.printStackTrace();
//				}
			}

		}
	}
}


第三钟方式：Condition和lock

    借助于Condition对象,我们也可以实现类似于wait()/notify()机制的功能,ReentrantLock重入锁相当于synchronized,每个Lock可以有任意数量的Condition对象,线程可以注册指定的Condition对象上,从而可以有选择性的进行线程通知,实现线程多路通知,在线程调度上更具有灵活性,在Condiiton对象的await()方法相当于wait()方法,使线程放弃锁,退出临界区进入等待状态,singal()/singalAll()方法相当于notify()/notifyAll()方法,唤醒处于等待状态的线程。

package com.demo.lock1;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 开启两个线程,一个线程向一个队列吐数据,当队列数据达到一定条数量的时候,通知另一个线程来消费数据
 * 
 * @author jiangry01
 *
 */
public class WriteAndRead {

	public static int THRESHOLD = 5;
	public static ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(5);
	public static volatile int COUNT;// 队列的容量
	public static ReentrantLock lock = new ReentrantLock();
	public Condition condition1 = lock.newCondition();
	public Condition condition2 = lock.newCondition();
	public Condition condition3 = lock.newCondition();

	public static void main(String[] args) throws InterruptedException {
		final WriteAndRead wr = new WriteAndRead();

		new Thread() {
			public void run() {
				while (true)
					wr.put();
			}
		}.start();

		new Thread() {
			public void run() {
				while (true)
					wr.get();
			}
		}.start();
	}

	// 写线程写入数据
	public void put() {
		// while (true){
		System.out.println("-----------");
		lock.lock();
		try {
			while (COUNT < THRESHOLD) {
				// 如果消息队列的消息数量
				System.out.println(Thread.currentThread().getName() + "写入： ");
				queue.put(0);
				COUNT = queue.size();
			}
			condition2.signal();
			condition1.await();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
			System.out.println(Thread.currentThread().getName() + "释放锁");
		}
	}

	// }
	// 读线程读取数据
	public void get() {
		// while (true){
		lock.lock();
		try {
			while (COUNT > 0) {
				// 如果消息队列的消息数量
				int i = queue.poll();
				System.out.println(Thread.currentThread().getName() + "读取： " + i);
				COUNT = queue.size();
				Thread.sleep(1000);
			}
			condition1.signal();
			condition2.await();
			System.out.println(queue.size());
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
	}
	// }
}


 

 

 

 

 

 

 

 

1、Netlink 机制简介
为了完成内核空间与用户空间通信，Linux提供了基于socket的Netlink通信机制，可以实现内核与用户空间数据的及时交换。Netlink是基于socket的通信机制，由于socket本身的双共性、突发性、不阻塞特点，因此能够很好的满足内核与用户空间小量数据的及时交互，因此在Linux 2.6内核中广泛使用，例如SELinux，Linux系统的防火墙分为内核态的netfilter和用户态的iptables，netfilter与iptables的数据交换就是通过Netlink机制完成。
netlink作为一种用户空间和内核空间通信的机制，它不光为了内核和用户通信，还可以作为IPC机制进行进程间通信。它也用到了sk_buff结构体，和网络套接字一样，更好的事情是它并没 有触及sk_buff里面的标准字段，而仅仅用了一个扩展的cb字段，cb在 sk_buff里面的定义是char cb[40];在netlink模块里面NETLINK_CB宏就是取cb字段的，也就是netlink所用的私有字段，这样的话你就可以用 netlink向任何执行实体传输任何数据了，不限于本机。

Linux操作系统中当CPU处于内核状态时，可以分为有用户上下文的状态和执行硬件、软件中断两种。其中当处于有用户上下文时，由于内核态和用户态的内 存映射机制不同，不可直接将本地变量传给用户态的内存区；处于硬件、软件中断时，无法直接向用户内存区传递数据，代码执行不可中断。针对传统的进程间通信 机制，他们均无法直接在内核态和用户态之间使用，原因如下：

通信方法                             无法介于内核态与用户态的原因

管道（不包括命名管道） 局限于父子进程间的通信。

消息队列                             在硬、软中断中无法无阻塞地接收数据。

信号量                                无法介于内核态和用户态使用。

内存共享                            需要信号量辅助，而信号量又无法使用。

套接字                                在硬、软中断中无法无阻塞地接收数据。

Netlink相对于其他的通信机制具有以下优点：

（1）使用Netlink通过自定义一种新的协议并加入协议族即可通过socket API使用Netlink协议完成数据交换，而ioctl和proc文件系统均需要通过程序加入相应的设备或文件。

（2）Netlink使用socket缓存队列，是一种异步通信机制，而ioctl是同步通信机制，如果传输的数据量较大，会影响系统性能。

（3）Netlink支持多播，属于一个Netlink组的模块和进程都能获得该多播消息。

（4）Netlink允许内核发起会话，而ioctl和系统调用只能由用户空间进程发起。

2、源码分析

2.1 linux\netlink.h

2.1.1 netlink 协议类型
#define NETLINK_ROUTE		0	/* Routing/device hook				*/
#define NETLINK_UNUSED		1	/* Unused number				*/
#define NETLINK_USERSOCK	2	/* Reserved for user mode socket protocols 	*/
#define NETLINK_FIREWALL	3	/* Firewalling hook				*/
#define NETLINK_INET_DIAG	4	/* INET socket monitoring			*/
#define NETLINK_NFLOG		5	/* netfilter/iptables ULOG */
#define NETLINK_XFRM		6	/* ipsec */
#define NETLINK_SELINUX		7	/* SELinux event notifications */
#define NETLINK_ISCSI		8	/* Open-iSCSI */
#define NETLINK_AUDIT		9	/* auditing */
#define NETLINK_FIB_LOOKUP	10	
#define NETLINK_CONNECTOR	11
#define NETLINK_NETFILTER	12	/* netfilter subsystem */
#define NETLINK_IP6_FW		13
#define NETLINK_DNRTMSG		14	/* DECnet routing messages */
#define NETLINK_KOBJECT_UEVENT	15	/* Kernel messages to userspace */
#define NETLINK_GENERIC		16
/* leave room for NETLINK_DM (DM Events) */
#define NETLINK_SCSITRANSPORT	18	/* SCSI Transports */
#define NETLINK_ECRYPTFS	19


可以根据自己的应用，创建新的通信协议。新协议的定义不能和内核已经定义的冲突，同时不能超过MAX_LINKS这个宏的限定，MAX_LINKS = 32。
2.1.2 netlink 地址格式
struct sockaddr_nl
{
	sa_family_t	nl_family;	/* AF_NETLINK	*/
	unsigned short	nl_pad;		/* zero		*/
	__u32		nl_pid;		/* port ID	*/
       	__u32		nl_groups;	/* multicast groups mask */
};

nl_pad 当前没有使用，因此要总是设置为 0，

nl_pid就是一个约定的通信端口，用户态使用的时候需要用一个非0的数字，一般来说可以直接采用上层应用的进程ID（未必是进程 ID,它只是用于区分不同的接收者或发送者的一个标识，用户可以根据自己需要设置该字段，只要是系统中不冲突的一个数字即可）。对于内核的地址，该值必须用0，即上层通过sendto向内核发送netlink消息，peer addr中nl_pid必须填写0。

nl_groups用于一个消息同时分发给不同的接收者，是一种组播应用。绑定时用于指定绑定者所要加入的多播组，这样绑定者就可以接收多播消息，发送 消息时可以用于指定多播组，这样就可以将消息发给多个接收者。这里nl_groups 为32位的无符号整形，所以可以指定32个多播组，每个进程可以加入多个多播组， 因为多播组是通过“或”操作，如果设置为 0，表示调用者不加入任何多播组。

本质上，nl_pid就是netlink的通信地址。除了通信地址，netlink还提供“协议”来标示通信实体，在创建socket的时候，需要指定netlink的通信协议号。每个协议号代表一种“应用”，上层可以用内核已经定义的协议和内核进行通信，获得内核已经提供的信息。netlink采用“协议号 + 通信端口”的方式构建自己的地址体系。


2.1.3 netlink 头部信息
struct nlmsghdr
{
	__u32		nlmsg_len;	/* Length of message including header */
	__u16		nlmsg_type;	/* Message content */
	__u16		nlmsg_flags;	/* Additional flags */
	__u32		nlmsg_seq;	/* Sequence number */
	__u32		nlmsg_pid;	/* Sending process port ID */
};


nlmsg_len 指定消息的总长度，包括紧跟该结构的数据部分长度以及该结构的大小。

nlmsg_type 用于应用内部定义消息的类型，它对netlink 内核实现是透明的，因此大部分情况下设置为 0。

nlmsg_seq 和 nlmsg_pid 用于应用追踪消息，前者表示顺序号，后者为消息来源进程 ID。其中pid是Netlink分配的一个ID，不同的值代表不同的socket通道，默认的值是进程PID。在某些情况下，这个值被设置为0，比如消息来自内核空间，或者想要Netlink来设置这个值。

nlmsg_flags 用于设置消息标志，可用的标志包括：

/* Flags values */
#define NLM_F_REQUEST		1	/* It is request message. 	*/
#define NLM_F_MULTI		2	/* Multipart message, terminated by NLMSG_DONE */
#define NLM_F_ACK		4	/* Reply with ack, with zero or error code */
#define NLM_F_ECHO		8	/* Echo this request 		*/

/* Modifiers to GET request */
#define NLM_F_ROOT	0x100	/* specify tree	root	*/
#define NLM_F_MATCH	0x200	/* return all matching	*/
#define NLM_F_ATOMIC	0x400	/* atomic GET		*/
#define NLM_F_DUMP	(NLM_F_ROOT|NLM_F_MATCH)

/* Modifiers to NEW request */
#define NLM_F_REPLACE	0x100	/* Override existing		*/
#define NLM_F_EXCL	0x200	/* Do not touch, if it exists	*/
#define NLM_F_CREATE	0x400	/* Create, if it does not exist	*/
#define NLM_F_APPEND	0x800	/* Add to end of list		*/

2.1.4 netlink 在skb_buff中的对应信息


struct netlink_skb_parms
{
	struct ucred		creds;		/* Skb credentials	*/
	__u32			pid;
	__u32			dst_group;
	kernel_cap_t		eff_cap;
	__u32			loginuid;	/* Login (audit) uid */
	__u32			sessionid;	/* Session id (audit) */
	__u32			sid;		/* SELinux security id */
};

内核态发送数据的时候，在skb_buff存储对应的发送地址等信息，供用户态需要时使用。
2.1.5 netlink 相关的宏
#define NLMSG_ALIGNTO	4
#define NLMSG_ALIGN(len) ( ((len)+NLMSG_ALIGNTO-1) & ~(NLMSG_ALIGNTO-1) )
#define NLMSG_HDRLEN	 ((int) NLMSG_ALIGN(sizeof(struct nlmsghdr)))
#define NLMSG_LENGTH(len) ((len)+NLMSG_ALIGN(NLMSG_HDRLEN))
#define NLMSG_SPACE(len) NLMSG_ALIGN(NLMSG_LENGTH(len))
#define NLMSG_DATA(nlh)  ((void*)(((char*)nlh) + NLMSG_LENGTH(0)))	//获取nlmsghdr头部之后的数据
#define NLMSG_NEXT(nlh,len)	 ((len) -= NLMSG_ALIGN((nlh)->nlmsg_len), \
				  (struct nlmsghdr*)(((char*)(nlh)) + NLMSG_ALIGN((nlh)->nlmsg_len)))
#define NLMSG_OK(nlh,len) ((len) >= (int)sizeof(struct nlmsghdr) && \
			   (nlh)->nlmsg_len >= sizeof(struct nlmsghdr) && \
			   (nlh)->nlmsg_len <= (len))
#define NLMSG_PAYLOAD(nlh,len) ((nlh)->nlmsg_len - NLMSG_SPACE((len)))


2.2 net\netlink\af_netlink.c
netlink的实现方法主要在 /net/netlink/af_netlink.c文件中。
2.2.1 netlink 套接字结构

struct netlink_sock {
	/* struct sock has to be the first member of netlink_sock */
	struct sock		sk;
	u32			pid;				//内核态pid
	u32			dst_pid;
	u32			dst_group;
	u32			flags;
	u32			subscriptions;
	u32			ngroups;			//组数
	unsigned long		*groups;		//组号
	unsigned long		state;
	wait_queue_head_t	wait;	//等待队列
	struct netlink_callback	*cb;
	struct mutex		*cb_mutex;
	struct mutex		cb_def_mutex;
	void			(*netlink_rcv)(struct sk_buff *skb);	//内核态接收用户态信息后的处理函数
	struct module		*module;
};


Groups ： 对于每一个netlink协议类型，可以使用多播的概念，最多可以有 32个多播组，每一个多播组用一个位表示，netlink 的多播特性使得发送消息给同一个组仅需要一次系统调用，因而对于需要多播消息的应用而言，大大地降低了系统调用的次数。

注意：用户态可以使用标准的socket APIs， socket(), bind(), sendmsg(), recvmsg() 和 close() 等函数就能很容易地使用 netlink socket，在用户空间可以直接通过socket函数来使用Netlink通信，区别是：


（1）netlink有自己的地址； 

创建socket，socket(PF_NETLINK, SOCK_DGRAM, NETLINK_XXX)；第一个参数必须是PF_NETLINK或者AF_NETLINK（Linux中实际为一个东西，它表示要使用netlink），第二个参数用SOCK_DGRAM和SOCK_RAW都没问题，第三个参数就是netlink的协议号。

（2）netlink接收到的消息带一个netlink自己的消息头；

用户态发送、接收的时候，需要在自己附带的消息前面要加上一个netlink的消息头：
struct tag_rcv_buf  
    {  
            struct nlmsghdr hdr;            //netlink的消息头  
            netlink_notify_s my_msg;        //通信实体消息  
    }st_snd_buf;  


2.2.2 创建netlink socket 
struct sock *
netlink_kernel_create(struct net *net, int unit, unsigned int groups,
		      void (*input)(struct sk_buff *skb),
		      struct mutex *cb_mutex, struct module *module)
{
	struct socket *sock;
	struct sock *sk;
	struct netlink_sock *nlk;
	unsigned long *listeners = NULL;

	BUG_ON(!nl_table);

	if (unit < 0 || unit >= MAX_LINKS)
		return NULL;

	if (sock_create_lite(PF_NETLINK, SOCK_DGRAM, unit, &sock))	//创建socket 结构
		return NULL;

	/*
	 * We have to just have a reference on the net from sk, but don't
	 * get_net it. Besides, we cannot get and then put the net here.
	 * So we create one inside init_net and the move it to net.
	 */

	if (__netlink_create(&init_net, sock, cb_mutex, unit) < 0)	//创建sock 结构
		goto out_sock_release_nosk;

	sk = sock->sk;
	sk_change_net(sk, net);

	if (groups < 32)
		groups = 32;

	listeners = kzalloc(NLGRPSZ(groups) + sizeof(struct listeners_rcu_head),
			    GFP_KERNEL);
	if (!listeners)
		goto out_sock_release;

	sk->sk_data_ready = netlink_data_ready;
	if (input)
		nlk_sk(sk)->netlink_rcv = input;	//设置内核接收Netlink消息的函数

	if (netlink_insert(sk, net, 0))
		goto out_sock_release;

	nlk = nlk_sk(sk);	//取得sock嵌入的netlink_sock结构体
	nlk->flags |= NETLINK_KERNEL_SOCKET;

	netlink_table_grab();
	if (!nl_table[unit].registered) {
		nl_table[unit].groups = groups;
		nl_table[unit].listeners = listeners;
		nl_table[unit].cb_mutex = cb_mutex;
		nl_table[unit].module = module;
		nl_table[unit].registered = 1;		//更新netlink_table结构体信息，每种协议对应一个netlink_table结构
	} else {
		kfree(listeners);
		nl_table[unit].registered++;
	}
	netlink_table_ungrab();
	return sk;

out_sock_release:
	kfree(listeners);
	netlink_kernel_release(sk);
	return NULL;

out_sock_release_nosk:
	sock_release(sock);
	return NULL;
}


内核态创建netlink socket的时候，调用 netlink_kernel_create() 函数。
2.2.3 内核态发送数据
内核态向用户态发送数据的时候，调用netlink_unicast() 函数。
int netlink_unicast(struct sock *ssk, struct sk_buff *skb,
		    u32 pid, int nonblock)
{
	struct sock *sk;
	int err;
	long timeo;

	skb = netlink_trim(skb, gfp_any());

	timeo = sock_sndtimeo(ssk, nonblock);
retry:
	sk = netlink_getsockbypid(ssk, pid); //找到内核中对应该pid和SOCK_DGRAM协议的sock结构，挂载在哈希表中
	if (IS_ERR(sk)) {
		kfree_skb(skb);
		return PTR_ERR(sk);
	}
	if (netlink_is_kernel(sk))	//若另一端也为内核端，调用其netlink_sock的input函数接受
		return netlink_unicast_kernel(sk, skb);
	if (sk_filter(sk, skb)) {
		err = skb->len;
		kfree_skb(skb);
		sock_put(sk);
		return err;
	}

	err = netlink_attachskb(sk, skb, &timeo, ssk);
	if (err == 1)
		goto retry;
	if (err)
		return err;

	return netlink_sendskb(sk, skb);	//挂载到用户态sock结构的receive_queue队列中， 然后调用sk_data_ready函数，唤醒用户态睡眠函数
}


2.2.4 内核态接收数据

在netlink_kernel_create函数中，可以设置接收函数。

2.2.5 用户态发送数据
static int netlink_sendmsg(struct kiocb *kiocb, struct socket *sock,
			   struct msghdr *msg, size_t len)
{
	struct sock_iocb *siocb = kiocb_to_siocb(kiocb);
	struct sock *sk = sock->sk;
	struct netlink_sock *nlk = nlk_sk(sk);
	struct sockaddr_nl *addr = msg->msg_name;	//目的地址信息
	u32 dst_pid;
	u32 dst_group;
	struct sk_buff *skb;
	int err;
	struct scm_cookie scm;

	if (msg->msg_flags&MSG_OOB)
		return -EOPNOTSUPP;

	if (NULL == siocb->scm)
		siocb->scm = &scm;
	err = scm_send(sock, msg, siocb->scm);
	if (err < 0)
		return err;

	if (msg->msg_namelen) {
		if (addr->nl_family != AF_NETLINK)
			return -EINVAL;
		dst_pid = addr->nl_pid;
		dst_group = ffs(addr->nl_groups);
		if ((dst_group || dst_pid) &&
		    !netlink_capable(sock, NL_NONROOT_SEND))
			return -EPERM;
	} else {
		dst_pid = nlk->dst_pid;
		dst_group = nlk->dst_group;
	}

	if (!nlk->pid) {
		err = netlink_autobind(sock);
		if (err)
			goto out;
	}

	err = -EMSGSIZE;
	if (len > sk->sk_sndbuf - 32)
		goto out;
	err = -ENOBUFS;
	skb = alloc_skb(len, GFP_KERNEL);		//分配一个sk_buff结构，目的是将msghdr结构转化为sk_buff结构
	if (skb == NULL)
		goto out;

	NETLINK_CB(skb).pid	= nlk->pid;	//本地的pid信息
	NETLINK_CB(skb).dst_group = dst_group;
	NETLINK_CB(skb).loginuid = audit_get_loginuid(current);
	NETLINK_CB(skb).sessionid = audit_get_sessionid(current);
	security_task_getsecid(current, &(NETLINK_CB(skb).sid));
	memcpy(NETLINK_CREDS(skb), &siocb->scm->creds, sizeof(struct ucred));

	/* What can I do? Netlink is asynchronous, so that
	   we will have to save current capabilities to
	   check them, when this message will be delivered
	   to corresponding kernel module.   --ANK (980802)
	 */

	err = -EFAULT;
	if (memcpy_fromiovec(skb_put(skb, len), msg->msg_iov, len)) {	//将数据拷贝到sk_buff中
		kfree_skb(skb);
		goto out;
	}

	err = security_netlink_send(sk, skb);
	if (err) {
		kfree_skb(skb);
		goto out;
	}

	if (dst_group) {
		atomic_inc(&skb->users);
		netlink_broadcast(sk, skb, dst_pid, dst_group, GFP_KERNEL);
	}
	err = netlink_unicast(sk, skb, dst_pid, msg->msg_flags&MSG_DONTWAIT);

out:
	return err;
}



2.2.6 用户态接收数据
static int netlink_recvmsg(struct kiocb *kiocb, struct socket *sock,
			   struct msghdr *msg, size_t len,
			   int flags)
{
	struct sock_iocb *siocb = kiocb_to_siocb(kiocb);
	struct scm_cookie scm;
	struct sock *sk = sock->sk;
	struct netlink_sock *nlk = nlk_sk(sk);
	int noblock = flags&MSG_DONTWAIT;
	size_t copied;
	struct sk_buff *skb, *data_skb;
	int err;

	if (flags&MSG_OOB)
		return -EOPNOTSUPP;

	copied = 0;

	skb = skb_recv_datagram(sk, flags, noblock, &err);	//从等待队列中接受一个数据包
	if (skb == NULL)
		goto out;

	data_skb = skb;

#ifdef CONFIG_COMPAT_NETLINK_MESSAGES
	if (unlikely(skb_shinfo(skb)->frag_list)) {
		/*
		 * If this skb has a frag_list, then here that means that we
		 * will have to use the frag_list skb's data for compat tasks
		 * and the regular skb's data for normal (non-compat) tasks.
		 *
		 * If we need to send the compat skb, assign it to the
		 * 'data_skb' variable so that it will be used below for data
		 * copying. We keep 'skb' for everything else, including
		 * freeing both later.
		 */
		if (flags & MSG_CMSG_COMPAT)
			data_skb = skb_shinfo(skb)->frag_list;
	}
#endif

	copied = data_skb->len;	//接收的数据包长度
	if (len < copied) {
		msg->msg_flags |= MSG_TRUNC;
		copied = len;
	}

	skb_reset_transport_header(data_skb);	//将数据包从sk_buff中拷贝到msghdr结构中
	err = skb_copy_datagram_iovec(data_skb, 0, msg->msg_iov, copied);

	if (msg->msg_name) {	//如果要求得到kernel的struct netlink_nl结构
		struct sockaddr_nl *addr = (struct sockaddr_nl *)msg->msg_name;
		addr->nl_family = AF_NETLINK;
		addr->nl_pad    = 0;
		addr->nl_pid	= NETLINK_CB(skb).pid;	//内核态pid
		addr->nl_groups	= netlink_group_mask(NETLINK_CB(skb).dst_group);	//发送分组
		msg->msg_namelen = sizeof(*addr);
	}

	if (nlk->flags & NETLINK_RECV_PKTINFO)
		netlink_cmsg_recv_pktinfo(msg, skb);

	if (NULL == siocb->scm) {
		memset(&scm, 0, sizeof(scm));
		siocb->scm = &scm;
	}
	siocb->scm->creds = *NETLINK_CREDS(skb);
	if (flags & MSG_TRUNC)
		copied = data_skb->len;

	skb_free_datagram(sk, skb);	//释放sk_buff数据包信息

	if (nlk->cb && atomic_read(&sk->sk_rmem_alloc) <= sk->sk_rcvbuf / 2)
		netlink_dump(sk);

	scm_recv(sock, msg, siocb->scm, flags);
out:
	netlink_rcv_wake(sk);	//接受唤醒
	return err ? : copied;
}


内核态将数据发送到用户态对应的sock的sk_receive_queue中，并且唤醒睡眠的进程。
3、测试

3.1 用户态与内核态通信

3.2.1 内核代码实现
#include <linux/init.h>
#include <linux/module.h>
#include <linux/timer.h>
#include <linux/time.h>
#include <linux/types.h>
#include <net/sock.h>
#include <net/netlink.h>

#define NETLINK_TEST 25
#define MAX_MSGSIZE 1024
int stringlength(char *s);
void sendnlmsg(char * message);
int pid;
int err;
struct sock *nl_sk = NULL;
int flag = 0;

void sendnlmsg(char *message)
{
    struct sk_buff *skb_1;
    struct nlmsghdr *nlh;
    int len = NLMSG_SPACE(MAX_MSGSIZE);
int slen = 0;
    if(!message || !nl_sk)
    {
        return ;
    }
    skb_1 = alloc_skb(len,GFP_KERNEL);
    if(!skb_1)
    {
        printk(KERN_ERR "my_net_link:alloc_skb_1 error\n");
    }
    slen = stringlength(message);
    nlh = nlmsg_put(skb_1,0,0,0,MAX_MSGSIZE,0);

    NETLINK_CB(skb_1).pid = 0;
    NETLINK_CB(skb_1).dst_group = 0;

    message[slen]= '\0';
    memcpy(NLMSG_DATA(nlh),message,slen+1);
    printk("my_net_link:send message '%s'.\n",(char *)NLMSG_DATA(nlh));

    netlink_unicast(nl_sk,skb_1,pid,MSG_DONTWAIT);
}

int stringlength(char *s)
{
    int slen = 0;
    for(; *s; s++){
        slen++;
    }
    return slen;
}

void nl_data_ready(struct sk_buff *__skb)
 {
     struct sk_buff *skb;
     struct nlmsghdr *nlh;
     char str[100];
     struct completion cmpl;
     int i=10;
     skb = skb_get (__skb);
     if(skb->len >= NLMSG_SPACE(0))
     {
         nlh = nlmsg_hdr(skb);

         memcpy(str, NLMSG_DATA(nlh), sizeof(str));
         printk("Message received:%s\n",str) ;
         pid = nlh->nlmsg_pid;
         while(i--)
         {
                init_completion(&cmpl);
                wait_for_completion_timeout(&cmpl,3 * HZ);
                sendnlmsg("I am from kernel!");
         }
         flag = 1;
         kfree_skb(skb);
     }

 }
int netlink_init(void)
{


    nl_sk = netlink_kernel_create(&init_net, NETLINK_TEST, 0,
                                 nl_data_ready, NULL, THIS_MODULE);

    if(!nl_sk){
        printk(KERN_ERR "my_net_link: create netlink socket error.\n");
        return 1;
    }

    printk("my_net_link: create netlink socket ok.\n");


    return 0;
}

static void netlink_exit(void)
{
    if(nl_sk != NULL){
        sock_release(nl_sk->sk_socket);
    }

    printk("my_net_link: module exited.\n");
}

module_init(netlink_init);
module_exit(netlink_exit);

MODULE_AUTHOR("LRQ");
MODULE_LICENSE("GPL");


3.2.2 用户态代码实现
#include <sys/stat.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <string.h>
#include <asm/types.h>
#include <linux/netlink.h>
#include <linux/socket.h>
#include <errno.h>

#define NETLINK_TEST 25
#define MAX_PAYLOAD 1024 // maximum payload size

int main(int argc, char* argv[])
{
    int state;
    struct sockaddr_nl src_addr, dest_addr;
    struct nlmsghdr *nlh = NULL;
    struct iovec iov;
    struct msghdr msg;
    int sock_fd, retval;
    int state_smg = 0;
    // Create a socket

    sock_fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_TEST);
    if(sock_fd == -1){
        printf("error getting socket: %s", strerror(errno));
        return -1;
}
// To prepare binding

    memset(&msg,0,sizeof(msg));
    memset(&src_addr, 0, sizeof(src_addr));
    src_addr.nl_family = AF_NETLINK;
    src_addr.nl_pid = getpid(); // self pid

    src_addr.nl_groups = 0; // multi cast


    retval = bind(sock_fd, (struct sockaddr*)&src_addr, sizeof(src_addr));
    if(retval < 0){
        printf("bind failed: %s", strerror(errno));
        close(sock_fd);
        return -1;
    }

    // To prepare recvmsg

nlh = (struct nlmsghdr *)malloc(NLMSG_SPACE(MAX_PAYLOAD));
    if(!nlh){
        printf("malloc nlmsghdr error!\n");
        close(sock_fd);
        return -1;
    }

    memset(&dest_addr,0,sizeof(dest_addr));
    dest_addr.nl_family = AF_NETLINK;
    dest_addr.nl_pid = 0;
    dest_addr.nl_groups = 0;

    nlh->nlmsg_len = NLMSG_SPACE(MAX_PAYLOAD);
    nlh->nlmsg_pid = getpid();
    nlh->nlmsg_flags = 0;
    strcpy(NLMSG_DATA(nlh),"Hello!");

    iov.iov_base = (void *)nlh;
    iov.iov_len = NLMSG_SPACE(MAX_PAYLOAD);
// iov.iov_len = nlh->nlmsg_len;

    memset(&msg, 0, sizeof(msg));

    msg.msg_name = (void *)&dest_addr;
    msg.msg_namelen = sizeof(dest_addr);
    msg.msg_iov = &iov;
    msg.msg_iovlen = 1;

    printf("state_smg\n");
    state_smg = sendmsg(sock_fd,&msg,0);

    if(state_smg == -1)
    {
        printf("get error sendmsg = %s\n",strerror(errno));
    }

    memset(nlh,0,NLMSG_SPACE(MAX_PAYLOAD));
    printf("waiting received!\n");
// Read message from kernel
    while(1){
        printf("In while recvmsg\n");
        state = recvmsg(sock_fd, &msg, 0);
        if(state<0)
        {
            printf("state<1");
        }
        printf("In while\n");
        printf("Received message: %s\n",(char *) NLMSG_DATA(nlh));
    }

    close(sock_fd);

    return 0;
}