[Flink流处理事件模式定义]的搜索结果

...tMQ实现发布/订阅模式。 二、什么是发布/订阅模式？ 发布/订阅模式是一种软件设计模式，主要用于处理事件驱动的应用程序。在这种模式下，咱们可以这么理解：生产者，也可以叫它“发布君”，它的工作就是往一个特定的“消息中心”——也就是主题或者交换机那儿发送消息。而消费者呢，换个接地气的名字就是“订阅达人”，它们会先关注这个“消息中心”。这样一来，只要“发布君”有新消息发出，“订阅达人”就能第一时间接收到所有这些消息啦！ 三、如何在RabbitMQ中实现发布/订阅模式？ 在RabbitMQ中，我们可以通过以下几个步骤来实现发布/订阅模式： 1. 创建并配置RabbitMQ环境 首先，我们需要在本地安装RabbitMQ，并启动服务。启动后，我们可以使用管理控制台查看RabbitMQ的状态和信息。 2. 创建交换机和队列 在RabbitMQ中，交换机和队列是两个基本的概念。交换机负责路由消息，而队列则用于存储消息。在接下来这一步，咱要做的是构建一个直通交换机和两个队列。其中一个队列呢，是专门用来接住生产者发过来的消息；另一个队列呢，则是用来给消费者传递他们的回复消息滴。 3. 编写生产者代码 在生产者代码中，我们将通过RabbitMQ的客户端API发送消息。首先，咱们得先捯饬出一个连接和通道，就像是搭起一座桥，然后像变魔术一样整出一个交换机，再配上两个队列，这两个队列就想象成是咱的消息暂存站。最后一步，就是把消息往这个交换机上一放，就像把信投进邮筒那样，完成发布啦！ python import pika 创建连接和通道 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() 创建交换机和队列 channel.exchange_declare(exchange='direct_logs', exchange_type='direct') 发布消息到交换机上 routing_key = 'INFO' message = "This is an info message" channel.basic_publish(exchange='direct_logs', routing_key=routing_key, body=message) print(" [x] Sent %r" % message) 关闭连接和通道 connection.close() 4. 编写消费者代码 在消费者代码中，我们将通过RabbitMQ的客户端API接收消息。首先，咱们得先搭起一座桥梁，建立起一条通道。然后，把队列和交换机牢牢地绑在一起。最后，从队列里取出消息，好好地“享用”一番。 python import pika 创建连接和通道 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) channel = connection.channel() 绑定队列到交换机上 queue_name = 'log_queue' channel.queue_bind(queue=queue_name, exchange='direct_logs', routing_key='INFO') 消费消息 def callback(ch, method, properties, body): print(" [x] Received %r" % body) channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True) 启动消费者 print(' [] Waiting for logs. To exit press CTRL+C') channel.start_consuming() 5. 运行代码并观察结果 现在，我们已经编写好了生产者和消费者的代码，接下来只需要运行这两个脚本就可以观察到发布/订阅模式的效果了。当生产者发送一条消息时，消费者会立即接收到这条消息，并打印出来。 四、总结 通过以上步骤，我们成功地在RabbitMQ中实现了发布/订阅模式。这简直就是个超级实用的编程模型，特别是在那些复杂的分布式系统里头，它能神奇地让不同应用程序之间的交流变得松耦合，这样一来，整个系统的稳定性和可靠性嗖嗖往上涨，就像给系统吃了颗定心丸一样。

...。 C 中的委托和事件(详解) 转载：http://www.cnblogs.com/SkySoot/archive/2012/04/05/2433639.html C 中的委托和事件 委托和事件在 .NET Framework 中的应用非常广泛，然而，较好地理解委托和事件对很多接触 C 时间不长的人来说并不容易。它们就像是一道槛儿，过了这个槛的人，觉得真是太容易了，而没有过去的人每次见到委托和事件就觉得心里堵得慌，浑身不自在。本章中，我将由浅入深地讲述什么是委托、为什么要使用委托、事件的由来、.NET Framework 中的委托和事件、委托中方法异常和超时的处理、委托与异步编程、委托和事件对Observer 设计模式的意义，对它们的编译代码也做了讨论。 1.1 理解委托 1.1.1 将方法作为方法的参数 我们先不管这个标题如何的绕口，也不管委托究竟是个什么东西，来看下面这两个最简单的方法，它们不过是在屏幕上输出一句问候的话语： public void GreetPeople(string name){EnglishGreeting(name);}public void EnglishGreeting(string name){Console.WriteLine("Good Morning, " + name);} 暂且不管这两个方法有没有什么实际意义。GreetPeople 用于向某人问好，当我们传递代表某人姓名的 name 参数，比如说“Liker”进去的时候，在这个方法中，将调用 EnglishGreeting 方法，再次传递 name 参数，EnglishGreeting 则用于向屏幕输出 “Good Morning, Liker”。 现在假设这个程序需要进行全球化，哎呀，不好了，我是中国人，我不明白“Good Morning”是什么意思，怎么办呢？好吧，我们再加个中文版的问候方法： public void ChineseGreeting(string name){Console.WriteLine("早上好, " + name);} 这时候，GreetPeople 也需要改一改了，不然如何判断到底用哪个版本的 Greeting 问候方法合适呢？在进行这个之前，我们最好再定义一个枚举作为判断的依据： public enum Language{English, Chinese}public void GreetPeople(string name, Language lang){switch (lang){case Language.English:EnglishGreeting(name);break;case Language.Chinese:ChineseGreeting(name);break;} } OK，尽管这样解决了问题，但我不说大家也很容易想到，这个解决方案的可扩展性很差，如果日后我们需要再添加韩文版、日文版，就不得不反复修改枚举和GreetPeople() 方法，以适应新的需求。 在考虑新的解决方案之前，我们先看看 GreetPeople 的方法签名： public void GreetPeople(string name, Language lang); 我们仅看 string name，在这里，string 是参数类型，name 是参数变量，当我们赋给 name 字符串“Liker”时，它就代表“Liker”这个值；当我们赋给它“李志中”时，它又代表着“李志中”这个值。然后，我们可以在方法体内对这个 name 进行其他操作。哎，这简直是废话么，刚学程序就知道了。 如果你再仔细想想，假如 GreetPeople() 方法可以接受一个参数变量，这个变量可以代表另一个方法，当我们给这个变量赋值 EnglishGreeting 的时候，它代表着 EnglsihGreeting() 这个方法；当我们给它赋值ChineseGreeting 的时候，它又代表着 ChineseGreeting() 法。我们将这个参数变量命名为 MakeGreeting，那么不是可以如同给 name 赋值时一样，在调用 GreetPeople()方法的时候，给这个MakeGreeting 参数也赋上值么(ChineseGreeting 或者EnglsihGreeting 等)？然后，我们在方法体内，也可以像使用别的参数一样使用MakeGreeting。但是，由于 MakeGreeting 代表着一个方法，它的使用方式应该和它被赋的方法(比如ChineseGreeting)是一样的，比如：MakeGreeting(name); 好了，有了思路了，我们现在就来改改GreetPeople()方法，那么它应该是这个样子了： public void GreetPeople(string name, MakeGreeting) { MakeGreeting(name); } 注意到 ，这个位置通常放置的应该是参数的类型，但到目前为止，我们仅仅是想到应该有个可以代表方法的参数，并按这个思路去改写 GreetPeople 方法，现在就出现了一个大问题：这个代表着方法的 MakeGreeting 参数应该是什么类型的？ 说明：这里已不再需要枚举了，因为在给MakeGreeting 赋值的时候动态地决定使用哪个方法，是 ChineseGreeting 还是 EnglishGreeting，而在这个两个方法内部，已经对使用“Good Morning”还是“早上好”作了区分。 聪明的你应该已经想到了，现在是委托该出场的时候了，但讲述委托之前，我们再看看MakeGreeting 参数所能代表的 ChineseGreeting()和EnglishGreeting()方法的签名： public void EnglishGreeting(string name) public void ChineseGreeting(string name) 如同 name 可以接受 String 类型的“true”和“1”，但不能接受bool 类型的true 和int 类型的1 一样。MakeGreeting 的参数类型定义应该能够确定 MakeGreeting 可以代表的方法种类，再进一步讲，就是 MakeGreeting 可以代表的方法的参数类型和返回类型。 于是，委托出现了：它定义了 MakeGreeting 参数所能代表的方法的种类，也就是 MakeGreeting 参数的类型。 本例中委托的定义： public delegate void GreetingDelegate(string name); 与上面 EnglishGreeting() 方法的签名对比一下，除了加入了delegate 关键字以外，其余的是不是完全一样？现在，让我们再次改动GreetPeople()方法，如下所示： public delegate void GreetingDelegate(string name);public void GreetPeople(string name, GreetingDelegate MakeGreeting){MakeGreeting(name);} 如你所见，委托 GreetingDelegate 出现的位置与 string 相同，string 是一个类型，那么 GreetingDelegate 应该也是一个类型，或者叫类(Class)。但是委托的声明方式和类却完全不同，这是怎么一回事？实际上，委托在编译的时候确实会编译成类。因为 Delegate 是一个类，所以在任何可以声明类的地方都可以声明委托。更多的内容将在下面讲述，现在，请看看这个范例的完整代码： public delegate void GreetingDelegate(string name);class Program{private static void EnglishGreeting(string name){Console.WriteLine("Good Morning, " + name);}private static void ChineseGreeting(string name){Console.WriteLine("早上好, " + name);}private static void GreetPeople(string name, GreetingDelegate MakeGreeting){MakeGreeting(name);}static void Main(string[] args){GreetPeople("Liker", EnglishGreeting);GreetPeople("李志中", ChineseGreeting);Console.ReadLine();} } 我们现在对委托做一个总结：委托是一个类，它定义了方法的类型，使得可以将方法当作另一个方法的参数来进行传递，这种将方法动态地赋给参数的做法，可以避免在程序中大量使用If … Else(Switch)语句，同时使得程序具有更好的可扩展性。 1.1.2 将方法绑定到委托 看到这里，是不是有那么点如梦初醒的感觉？于是，你是不是在想：在上面的例子中，我不一定要直接在 GreetPeople() 方法中给 name 参数赋值，我可以像这样使用变量： static void Main(string[] args){GreetPeople("Liker", EnglishGreeting);GreetPeople("李志中", ChineseGreeting);Console.ReadLine();} 而既然委托 GreetingDelegate 和类型 string 的地位一样，都是定义了一种参数类型，那么，我是不是也可以这么使用委托？ static void Main(string[] args){GreetingDelegate delegate1, delegate2;delegate1 = EnglishGreeting;delegate2 = ChineseGreeting;GreetPeople("Liker", delegate1);GreetPeople("李志中", delegate2);Console.ReadLine();} 如你所料，这样是没有问题的，程序一如预料的那样输出。这里，我想说的是委托不同于 string 的一个特性：可以将多个方法赋给同一个委托，或者叫将多个方法绑定到同一个委托，当调用这个委托的时候，将依次调用其所绑定的方法。在这个例子中，语法如下： static void Main(string[] args){GreetingDelegate delegate1;delegate1 = EnglishGreeting; delegate1 += ChineseGreeting;GreetPeople("Liker", delegate1);Console.ReadLine();} 实际上，我们可以也可以绕过GreetPeople 方法，通过委托来直接调用EnglishGreeting 和ChineseGreeting： static void Main(string[] args){GreetingDelegate delegate1;delegate1 = EnglishGreeting;delegate1 += ChineseGreeting; delegate1("Liker");Console.ReadLine();} 说明：这在本例中是没有问题的，但回头看下上面 GreetPeople() 的定义，在它之中可以做一些对于 EnglshihGreeting 和 ChineseGreeting 来说都需要进行的工作，为了简便我做了省略。 注意这里，第一次用的“=”，是赋值的语法；第二次，用的是“+=”，是绑定的语法。如果第一次就使用“+=”，将出现“使用了未赋值的局部变量”的编译错误。我们也可以使用下面的代码来这样简化这一过程： GreetingDelegate delegate1 = new GreetingDelegate(EnglishGreeting);delegate1 += ChineseGreeting; 既然给委托可以绑定一个方法，那么也应该有办法取消对方法的绑定，很容易想到，这个语法是“-=”： static void Main(string[] args){GreetingDelegate delegate1 = new GreetingDelegate(EnglishGreeting);delegate1 += ChineseGreeting;GreetPeople("Liker", delegate1);Console.WriteLine();delegate1 -= EnglishGreeting;GreetPeople("李志中", delegate1);Console.ReadLine();} 让我们再次对委托作个总结： 使用委托可以将多个方法绑定到同一个委托变量，当调用此变量时(这里用“调用”这个词，是因为此变量代表一个方法)，可以依次调用所有绑定的方法。 1.2 事件的由来 1.2.1 更好的封装性 我们继续思考上面的程序：上面的三个方法都定义在 Programe 类中，这样做是为了理解的方便，实际应用中，通常都是 GreetPeople 在一个类中，ChineseGreeting 和 EnglishGreeting 在另外的类中。现在你已经对委托有了初步了解，是时候对上面的例子做个改进了。假设我们将 GreetingPeople() 放在一个叫 GreetingManager 的类中，那么新程序应该是这个样子的： namespace Delegate{public delegate void GreetingDelegate(string name);public class GreetingManager{public void GreetPeople(string name, GreetingDelegate MakeGreeting){MakeGreeting(name);} }class Program{private static void EnglishGreeting(string name){Console.WriteLine("Good Morning, " + name);}private static void ChineseGreeting(string name){Console.WriteLine("早上好, " + name);}static void Main(string[] args){GreetingManager gm = new GreetingManager();gm.GreetPeople("Liker", EnglishGreeting);gm.GreetPeople("李志中", ChineseGreeting);} }} 我们运行这段代码，嗯，没有任何问题。程序一如预料地那样输出了： // Good Morning, Liker 早上好, 李志中 // 现在，假设我们需要使用上一节学到的知识，将多个方法绑定到同一个委托变量，该如何做呢？让我们再次改写代码： static void Main(string[] args){GreetingManager gm = new GreetingManager();GreetingDelegate delegate1;delegate1 = EnglishGreeting;delegate1 += ChineseGreeting;gm.GreetPeople("Liker", delegate1);} 输出： Good Morning, Liker 早上好, Liker 到了这里，我们不禁想到：面向对象设计，讲究的是对象的封装，既然可以声明委托类型的变量(在上例中是delegate1)，我们何不将这个变量封装到 GreetManager 类中？在这个类的客户端中使用不是更方便么？于是，我们改写GreetManager 类，像这样： public class GreetingManager{/// <summary>/// 在 GreetingManager 类的内部声明 delegate1 变量/// </summary>public GreetingDelegate delegate1;public void GreetPeople(string name, GreetingDelegate MakeGreeting){MakeGreeting(name);} } 现在，我们可以这样使用这个委托变量： static void Main(string[] args){GreetingManager gm = new GreetingManager();gm.delegate1 = EnglishGreeting;gm.delegate1 += ChineseGreeting;gm.GreetPeople("Liker", gm.delegate1);} 输出为： Good Morning, Liker 早上好, Liker 尽管这样做没有任何问题，但我们发现这条语句很奇怪。在调用gm.GreetPeople 方法的时候，再次传递了gm 的delegate1 字段, 既然如此，我们何不修改 GreetingManager 类成这样： public class GreetingManager{/// <summary>/// 在 GreetingManager 类的内部声明 delegate1 变量/// </summary>public GreetingDelegate delegate1;public void GreetPeople(string name){if (delegate1 != null) // 如果有方法注册委托变量{ delegate1(name); // 通过委托调用方法} }} 在客户端，调用看上去更简洁一些： static void Main(string[] args){GreetingManager gm = new GreetingManager();gm.delegate1 = EnglishGreeting;gm.delegate1 += ChineseGreeting;gm.GreetPeople("Liker"); //注意，这次不需要再传递 delegate1 变量} 尽管这样达到了我们要的效果，但是还是存在着问题：在这里，delegate1 和我们平时用的string 类型的变量没有什么分别，而我们知道，并不是所有的字段都应该声明成public，合适的做法是应该public 的时候public，应该private 的时候private。 我们先看看如果把 delegate1 声明为 private 会怎样？结果就是：这简直就是在搞笑。因为声明委托的目的就是为了把它暴露在类的客户端进行方法的注册，你把它声明为 private 了，客户端对它根本就不可见，那它还有什么用？ 再看看把delegate1 声明为 public 会怎样？结果就是：在客户端可以对它进行随意的赋值等操作，严重破坏对象的封装性。 最后，第一个方法注册用“=”，是赋值语法，因为要进行实例化，第二个方法注册则用的是“+=”。但是，不管是赋值还是注册，都是将方法绑定到委托上，除了调用时先后顺序不同，再没有任何的分别，这样不是让人觉得很别扭么？ 现在我们想想，如果delegate1 不是一个委托类型，而是一个string 类型，你会怎么做？答案是使用属性对字段进行封装。 于是，Event 出场了，它封装了委托类型的变量，使得：在类的内部，不管你声明它是public还是protected，它总是private 的。在类的外部，注册“+=”和注销“-=”的访问限定符与你在声明事件时使用的访问符相同。我们改写GreetingManager 类，它变成了这个样子： public class GreetingManager{//这一次我们在这里声明一个事件public event GreetingDelegate MakeGreet;public void GreetPeople(string name){MakeGreet(name);} } 很容易注意到：MakeGreet 事件的声明与之前委托变量 delegate1 的声明唯一的区别是多了一个 event 关键字。看到这里，在结合上面的讲解，你应该明白到：事件其实没什么不好理解的，声明一个事件不过类似于声明一个进行了封装的委托类型的变量而已。 为了证明上面的推论，如果我们像下面这样改写Main 方法： static void Main(string[] args){GreetingManager gm = new GreetingManager();gm.MakeGreet = EnglishGreeting; // 编译错误1gm.MakeGreet += ChineseGreeting;gm.GreetPeople("Liker");} 会得到编译错误： 1.2.2 限制类型能力 使用事件不仅能获得比委托更好的封装性以外，还能限制含有事件的类型的能力。这是什么意思呢？它的意思是说：事件应该由事件发布者触发，而不应该由事件的客户端（客户程序）来触发。请看下面的范例： using System;class Program{static void Main(string[] args){Publishser pub = new Publishser();Subscriber sub = new Subscriber();pub.NumberChanged += new NumberChangedEventHandler(sub.OnNumberChanged);pub.DoSomething(); // 应该通过DoSomething()来触发事件pub.NumberChanged(100); // 但可以被这样直接调用，对委托变量的不恰当使用} }/// <summary>/// 定义委托/// </summary>/// <param name="count"></param>public delegate void NumberChangedEventHandler(int count);/// <summary>/// 定义事件发布者/// </summary>public class Publishser{private int count;public NumberChangedEventHandler NumberChanged; // 声明委托变量//public event NumberChangedEventHandler NumberChanged; // 声明一个事件public void DoSomething(){// 在这里完成一些工作 ...if (NumberChanged != null) // 触发事件{ count++;NumberChanged(count);} }}/// <summary>/// 定义事件订阅者/// </summary>public class Subscriber{public void OnNumberChanged(int count){Console.WriteLine("Subscriber notified: count = {0}", count);} } 上面代码定义了一个NumberChangedEventHandler 委托，然后我们创建了事件的发布者Publisher 和订阅者Subscriber。当使用委托变量时，客户端可以直接通过委托变量触发事件，也就是直接调用pub.NumberChanged(100)，这将会影响到所有注册了该委托的订阅者。而事件的本意应该为在事件发布者在其本身的某个行为中触发，比如说在方法DoSomething()中满足某个条件后触发。通过添加event 关键字来发布事件，事件发布者的封装性会更好，事件仅仅是供其他类型订阅，而客户端不能直接触发事件（语句pub.NumberChanged(100)无法通过编译），事件只能在事件发布者Publisher 类的内部触发（比如在方法pub.DoSomething()中），换言之，就是NumberChanged(100)语句只能在Publisher 内部被调用。大家可以尝试一下，将委托变量的声明那行代码注释掉，然后取消下面事件声明的注释。此时程序是无法编译的，当你使用了event 关键字之后，直接在客户端触发事件这种行为，也就是直接调用pub.NumberChanged(100)，是被禁止的。事件只能通过调用DoSomething() 来触发。这样才是事件的本意，事件发布者的封装才会更好。 就好像如果我们要定义一个数字类型，我们会使用int 而不是使用object 一样，给予对象过多的能力并不见得是一件好事，应该是越合适越好。尽管直接使用委托变量通常不会有什么问题，但它给了客户端不应具有的能力，而使用事件，可以限制这一能力，更精确地对类型进行封装。 说 明：这里还有一个约定俗称的规定，就是订阅事件的方法的命名，通常为“On 事件名”，比如这里的OnNumberChanged。 1.3 委托的编译代码 这时候，我们注释掉编译错误的行，然后重新进行编译，再借助 Reflactor 来对 event 的声明语句做一探究，看看为什么会发生这样的错误： 可以看到，实际上尽管我们在GreetingManager 里将 MakeGreet 声明为public，但是，实际上MakeGreet 会被编译成私有字段，难怪会发生上面的编译错误了，因为它根本就不允许在GreetingManager 类的外面以赋值的方式访问，从而验证了我们上面所做的推论。 我们再进一步看下MakeGreet 所产生的代码： // private GreetingDelegate MakeGreet; //对事件的声明实际是声明一个私有的委托变量 [MethodImpl(MethodImplOptions.Synchronized)] public void add_MakeGreet(GreetingDelegate value) { this.MakeGreet = (GreetingDelegate) Delegate.Combine(this.MakeGreet, value); } [MethodImpl(MethodImplOptions.Synchronized)] public void remove_MakeGreet(GreetingDelegate value) { this.MakeGreet = (GreetingDelegate) Delegate.Remove(this.MakeGreet, value); } // 现在已经很明确了：MakeGreet 事件确实是一个GreetingDelegate 类型的委托，只不过不管是不是声明为public，它总是被声明为private。另外，它还有两个方法，分别是add_MakeGreet和remove_MakeGreet，这两个方法分别用于注册委托类型的方法和取消注册。实际上也就是：“+= ”对应 add_MakeGreet，“-=”对应remove_MakeGreet。而这两个方法的访问限制取决于声明事件时的访问限制符。 在add_MakeGreet()方法内部，实际上调用了System.Delegate 的Combine()静态方法，这个方法用于将当前的变量添加到委托链表中。 我们前面提到过两次，说委托实际上是一个类，在我们定义委托的时候： // public delegate void GreetingDelegate(string name); // 当编译器遇到这段代码的时候，会生成下面这样一个完整的类： // public class GreetingDelegate:System.MulticastDelegate { public GreetingDelegate(object @object, IntPtr method); public virtual IAsyncResult BeginInvoke(string name, AsyncCallback callback, object @object); public virtual void EndInvoke(IAsyncResult result); public virtual void Invoke(string name); } // 1.4 .NET 框架中的委托和事件 1.4.1 范例说明 上面的例子已不足以再进行下面的讲解了，我们来看一个新的范例，因为之前已经介绍了很多的内容，所以本节的进度会稍微快一些! 假设我们有个高档的热水器，我们给它通上电，当水温超过95 度的时候：1、扬声器会开始发出语音，告诉你水的温度；2、液晶屏也会改变水温的显示，来提示水已经快烧开了。 现在我们需要写个程序来模拟这个烧水的过程，我们将定义一个类来代表热水器，我们管它叫：Heater，它有代表水温的字段，叫做 temperature；当然，还有必不可少的给水加热方法 BoilWater()，一个发出语音警报的方法 MakeAlert()，一个显示水温的方法，ShowMsg()。 namespace Delegate{/// <summary>/// 热水器/// </summary>public class Heater{/// <summary>/// 水温/// </summary>private int temperature;/// <summary>/// 烧水/// </summary>public void BoilWater(){for (int i = 0; i <= 100; i++){temperature = i;if (temperature > 95){MakeAlert(temperature);ShowMsg(temperature);} }}/// <summary>/// 发出语音警报/// </summary>/// <param name="param"></param>private void MakeAlert(int param){Console.WriteLine("Alarm：嘀嘀嘀，水已经 {0} 度了：", param);}/// <summary>/// 显示水温/// </summary>/// <param name="param"></param>private void ShowMsg(int param){Console.WriteLine("Display：水快开了，当前温度：{0}度。", param);} }class Program{static void Main(){Heater ht = new Heater();ht.BoilWater();} }} 1.4.2 Observer 设计模式简介 上面的例子显然能完成我们之前描述的工作，但是却并不够好。现在假设热水器由三部分组成：热水器、警报器、显示器，它们来自于不同厂商并进行了组装。那么，应该是热水器仅仅负责烧水，它不能发出警报也不能显示水温；在水烧开时由警报器发出警报、显示器显示提示和水温。 这时候，上面的例子就应该变成这个样子： /// <summary>/// 热水器/// </summary>public class Heater{private int temperature; private void BoilWater(){for (int i = 0; i <= 100; i++){temperature = i;} }}/// <summary>/// 警报器/// </summary>public class Alarm{private void MakeAlert(int param){Console.WriteLine("Alarm：嘀嘀嘀，水已经 {0} 度了：", param);} }/// <summary>/// 显示器/// </summary>public class Display{private void ShowMsg(int param){Console.WriteLine("Display：水已烧开，当前温度：{0}度。", param);} } 这里就出现了一个问题：如何在水烧开的时候通知报警器和显示器？ 在继续进行之前，我们先了解一下Observer 设计模式，Observer 设计模式中主要包括如下两类对象： Subject：监视对象，它往往包含着其他对象所感兴趣的内容。在本范例中，热水器就是一个监视对象，它包含的其他对象所感兴趣的内容，就是 temprature 字段，当这个字段的值快到100 时，会不断把数据发给监视它的对象。 Observer：监视者，它监视Subject，当 Subject 中的某件事发生的时候，会告知Observer，而Observer 则会采取相应的行动。在本范例中，Observer 有警报器和显示器，它们采取的行动分别是发出警报和显示水温。 在本例中，事情发生的顺序应该是这样的： 1. 警报器和显示器告诉热水器，它对它的温度比较感兴趣(注册)。 2. 热水器知道后保留对警报器和显示器的引用。 3. 热水器进行烧水这一动作，当水温超过 95 度时，通过对警报器和显示器的引用，自动调用警报器的MakeAlert()方法、显示器的ShowMsg()方法。 类似这样的例子是很多的，GOF 对它进行了抽象，称为 Observer 设计模式：Observer 设计模式是为了定义对象间的一种一对多的依赖关系，以便于当一个对象的状态改变时，其他依赖于它的对象会被自动告知并更新。Observer 模式是一种松耦合的设计模式。 1.4.3 实现范例的Observer 设计模式 我们之前已经对委托和事件介绍很多了，现在写代码应该很容易了，现在在这里直接给出代码，并在注释中加以说明。 namespace Delegate{public class Heater{private int temperature;public delegate void BoilHandler(int param);public event BoilHandler BoilEvent;public void BoilWater(){for (int i = 0; i <= 100; i++){temperature = i;if (temperature > 95){if (BoilEvent != null){ BoilEvent(temperature); // 调用所有注册对象的方法} }} }}public class Alarm{public void MakeAlert(int param){Console.WriteLine("Alarm：嘀嘀嘀，水已经 {0} 度了：", param);} }public class Display{public static void ShowMsg(int param) // 静态方法{ Console.WriteLine("Display：水快烧开了，当前温度：{0}度。", param);} }class Program{static void Main(){Heater heater = new Heater();Alarm alarm = new Alarm();heater.BoilEvent += alarm.MakeAlert; // 注册方法heater.BoilEvent += (new Alarm()).MakeAlert; // 给匿名对象注册方法heater.BoilEvent += Display.ShowMsg; // 注册静态方法heater.BoilWater(); // 烧水，会自动调用注册过对象的方法} }} 输出为： // Alarm：嘀嘀嘀，水已经 96 度了： Alarm：嘀嘀嘀，水已经 96 度了： Display：水快烧开了，当前温度：96 度。 // 省略... // 1.4.4 .NET 框架中的委托与事件 尽管上面的范例很好地完成了我们想要完成的工作，但是我们不仅疑惑：为什么.NET Framework 中的事件模型和上面的不同？为什么有很多的EventArgs 参数？ 在回答上面的问题之前，我们先搞懂 .NET Framework 的编码规范： 1. 委托类型的名称都应该以 EventHandler 结束。 2. 委托的原型定义：有一个void 返回值，并接受两个输入参数：一个Object 类型，一个EventArgs 类型(或继承自EventArgs)。 3. 事件的命名为委托去掉 EventHandler 之后剩余的部分。 4. 继承自 EventArgs 的类型应该以EventArgs 结尾。 再做一下说明： 1. 委托声明原型中的Object 类型的参数代表了Subject，也就是监视对象，在本例中是Heater(热水器)。回调函数(比如Alarm 的MakeAlert)可以通过它访问触发事件的对象(Heater)。 2. EventArgs 对象包含了Observer 所感兴趣的数据，在本例中是temperature。 上面这些其实不仅仅是为了编码规范而已，这样也使得程序有更大的灵活性。比如说，如果我们不光想获得热水器的温度，还想在Observer 端(警报器或者显示器)方法中获得它的生产日期、型号、价格，那么委托和方法的声明都会变得很麻烦，而如果我们将热水器的引用传给警报器的方法，就可以在方法中直接访问热水器了。 现在我们改写之前的范例，让它符合.NET Framework的规范： using System;using System.Collections.Generic;using System.Text;namespace Delegate{public class Heater{private int temperature;public string type = "RealFire 001"; // 添加型号作为演示public string area = "China Xian"; // 添加产地作为演示public delegate void BoiledEventHandler(Object sender, BoiledEventArgs e);public event BoiledEventHandler Boiled; // 声明事件// 定义 BoiledEventArgs 类，传递给 Observer 所感兴趣的信息public class BoiledEventArgs : EventArgs{public readonly int temperature;public BoiledEventArgs(int temperature){this.temperature = temperature;} }// 可以供继承自 Heater 的类重写，以便继承类拒绝其他对象对它的监视protected virtual void OnBoiled(BoiledEventArgs e){if (Boiled != null){Boiled(this, e); // 调用所有注册对象的方法} }public void BoilWater(){for (int i = 0; i <= 100; i++){temperature = i;if (temperature > 95){// 建立BoiledEventArgs 对象。BoiledEventArgs e = new BoiledEventArgs(temperature);OnBoiled(e); // 调用 OnBolied 方法} }}public class Alarm{public void MakeAlert(Object sender, Heater.BoiledEventArgs e){Heater heater = (Heater)sender; // 这里是不是很熟悉呢？// 访问 sender 中的公共字段Console.WriteLine("Alarm：{0} - {1}: ", heater.area, heater.type);Console.WriteLine("Alarm: 嘀嘀嘀，水已经 {0} 度了：", e.temperature);Console.WriteLine();} }public class Display{public static void ShowMsg(Object sender, Heater.BoiledEventArgs e) // 静态方法{Heater heater = (Heater)sender;Console.WriteLine("Display：{0} - {1}: ", heater.area, heater.type);Console.WriteLine("Display：水快烧开了，当前温度：{0}度。", e.temperature);Console.WriteLine();} }class Program{static void Main(){Heater heater = new Heater();Alarm alarm = new Alarm();heater.Boiled += alarm.MakeAlert; //注册方法heater.Boiled += (new Alarm()).MakeAlert; //给匿名对象注册方法heater.Boiled += new Heater.BoiledEventHandler(alarm.MakeAlert); //也可以这么注册heater.Boiled += Display.ShowMsg; //注册静态方法heater.BoilWater(); //烧水，会自动调用注册过对象的方法} }} } 输出为： Alarm：China Xian - RealFire 001: Alarm: 嘀嘀嘀，水已经 96 度了： Alarm：China Xian - RealFire 001: Alarm: 嘀嘀嘀，水已经 96 度了： Alarm：China Xian - RealFire 001: Alarm: 嘀嘀嘀，水已经 96 度了： Display：China Xian - RealFire 001: Display：水快烧开了，当前温度：96 度。 // 省略 ... 1.5 委托进阶 1.5.1 为什么委托定义的返回值通常都为 void ？ 尽管并非必需，但是我们发现很多的委托定义返回值都为 void，为什么呢？这是因为委托变量可以供多个订阅者注册，如果定义了返回值，那么多个订阅者的方法都会向发布者返回数值，结果就是后面一个返回的方法值将前面的返回值覆盖掉了，因此，实际上只能获得最后一个方法调用的返回值。可以运行下面的代码测试一下。除此以外，发布者和订阅者是松耦合的，发布者根本不关心谁订阅了它的事件、为什么要订阅，更别说订阅者的返回值了，所以返回订阅者的方法返回值大多数情况下根本没有必要。 1.5.2 如何让事件只允许一个客户订阅？ 少数情况下，比如像上面，为了避免发生“值覆盖”的情况（更多是在异步调用方法时，后面会讨论），我们可能想限制只允许一个客户端注册。此时怎么做呢？我们可以向下面这样，将事件声明为private 的，然后提供两个方法来进行注册和取消注册： public class Publishser{private event GeneralEventHandler NumberChanged; // 声明一个私有事件// 注册事件public void Register(GeneralEventHandler method){NumberChanged = method;}// 取消注册public void UnRegister(GeneralEventHandler method){NumberChanged -= method;}public void DoSomething(){// 做某些其余的事情if (NumberChanged != null){ // 触发事件string rtn = NumberChanged();Console.WriteLine("Return: {0}", rtn); // 打印返回的字符串，输出为Subscriber3} }} 注意上面，在UnRegister()中，没有进行任何判断就使用了NumberChanged -= method 语句。这是因为即使method 方法没有进行过注册，此行语句也不会有任何问题，不会抛出异常，仅仅是不会产生任何效果而已。 注意在Register()方法中，我们使用了赋值操作符“=”，而非“+=”，通过这种方式就避免了多个方法注册。 1.7 委托和方法的异步调用 通常情况下，如果需要异步执行一个耗时的操作，我们会新起一个线程，然后让这个线程去执行代码。但是对于每一个异步调用都通过创建线程来进行操作显然会对性能产生一定的影响，同时操作也相对繁琐一些。.NET 中可以通过委托进行方法的异步调用，就是说客户端在异步调用方法时，本身并不会因为方法的调用而中断，而是从线程池中抓取一个线程去执行该方法，自身线程（主线程）在完成抓取线程这一过程之后，继续执行下面的代码，这样就实现了代码的并行执行。使用线程池的好处就是避免了频繁进行异步调用时创建、销毁线程的开销。当我们在委托对象上调用BeginInvoke()时，便进行了一个异步的方法调用。 事件发布者和订阅者之间往往是松耦合的，发布者通常不需要获得订阅者方法执行的情况；而当使用异步调用时，更多情况下是为了提升系统的性能，而并非专用于事件的发布和订阅这一编程模型。而在这种情况下使用异步编程时，就需要进行更多的控制，比如当异步执行方法的方法结束时通知客户端、返回异步执行方法的返回值等。本节就对 BeginInvoke() 方法、EndInvoke() 方法和其相关的 IAysncResult 做一个简单的介绍。 我们先看这样一段代码，它演示了不使用异步调用的通常情况： class Program7{static void Main(string[] args){Console.WriteLine("Client application started!\n");Thread.CurrentThread.Name = "Main Thread";Calculator cal = new Calculator();int result = cal.Add(2, 5);Console.WriteLine("Result: {0}\n", result);// 做某些其它的事情，模拟需要执行3 秒钟for (int i = 1; i <= 3; i++){Thread.Sleep(TimeSpan.FromSeconds(i));Console.WriteLine("{0}: Client executed {1} second(s).", Thread.CurrentThread.Name, i);}Console.WriteLine("\nPress any key to exit...");Console.ReadLine();} }public class Calculator{public int Add(int x, int y){if (Thread.CurrentThread.IsThreadPoolThread){Thread.CurrentThread.Name = "Pool Thread";}Console.WriteLine("Method invoked!");// 执行某些事情，模拟需要执行2 秒钟for (int i = 1; i <= 2; i++){Thread.Sleep(TimeSpan.FromSeconds(i));Console.WriteLine("{0}: Add executed {1} second(s).", Thread.CurrentThread.Name, i);}Console.WriteLine("Method complete!");return x + y;} } 上面代码有几个关于对于线程的操作，如果不了解可以看一下下面的说明，如果你已经了解可以直接跳过： 1. Thread.Sleep()，它会让执行当前代码的线程暂停一段时间（如果你对线程的概念比较陌生，可以理解为使程序的执行暂停一段时间），以毫秒为单位，比如Thread.Sleep(1000)，将会使线程暂停1 秒钟。在上面我使用了它的重载方法，个人觉得使用TimeSpan.FromSeconds(1)，可读性更好一些。 2. Thread.CurrentThread.Name，通过这个属性可以设置、获取执行当前代码的线程的名称，值得注意的是这个属性只可以设置一次，如果设置两次，会抛出异常。 3. Thread.IsThreadPoolThread，可以判断执行当前代码的线程是否为线程池中的线程。 通过这几个方法和属性，有助于我们更好地调试异步调用方法。上面代码中除了加入了一些对线程的操作以外再没有什么特别之处。我们建了一个Calculator 类，它只有一个Add 方法，我们模拟了这个方法需要执行2 秒钟时间，并且每隔一秒进行一次输出。而在客户端程序中，我们使用result 变量保存了方法的返回值并进行了打印。随后，我们再次模拟了客户端程序接下来的操作需要执行2 秒钟时间。运行这段程序，会产生下面的输出： // Client application started! Method invoked! Main Thread: Add executed 1 second(s). Main Thread: Add executed 2 second(s). Method complete! Result: 7 Main Thread: Client executed 1 second(s). Main Thread: Client executed 2 second(s). Main Thread: Client executed 3 second(s). Press any key to exit... // 如果你确实执行了这段代码，会看到这些输出并不是一瞬间输出的，而是执行了大概5 秒钟的时间，因为线程是串行执行的，所以在执行完 Add() 方法之后才会继续客户端剩下的代码。 接下来我们定义一个AddDelegate 委托，并使用BeginInvoke()方法来异步地调用它。在上面已经介绍过，BeginInvoke()除了最后两个参数为AsyncCallback 类型和Object 类型以外，前面的参数类型和个数与委托定义相同。另外BeginInvoke()方法返回了一个实现了IAsyncResult 接口的对象（实际上就是一个AsyncResult 类型实例，注意这里IAsyncResult 和AysncResult 是不同的，它们均包含在.NET Framework 中）。 AsyncResult 的用途有这么几个：传递参数，它包含了对调用了BeginInvoke()的委托的引用；它还包含了BeginInvoke()的最后一个Object 类型的参数；它可以鉴别出是哪个方法的哪一次调用，因为通过同一个委托变量可以对同一个方法调用多次。 EndInvoke()方法接受IAsyncResult 类型的对象（以及ref 和out 类型参数，这里不讨论了，对它们的处理和返回值类似），所以在调用BeginInvoke()之后，我们需要保留IAsyncResult，以便在调用EndInvoke()时进行传递。这里最重要的就是EndInvoke()方法的返回值，它就是方法的返回值。除此以外，当客户端调用EndInvoke()时，如果异步调用的方法没有执行完毕，则会中断当前线程而去等待该方法，只有当异步方法执行完毕后才会继续执行后面的代码。所以在调用完BeginInvoke()后立即执行EndInvoke()是没有任何意义的。我们通常在尽可能早的时候调用BeginInvoke()，然后在需要方法的返回值的时候再去调用EndInvoke()，或者是根据情况在晚些时候调用。说了这么多，我们现在看一下使用异步调用改写后上面的代码吧： using System.Threading;using System;public delegate int AddDelegate(int x, int y);class Program8{static void Main(string[] args){Console.WriteLine("Client application started!\n");Thread.CurrentThread.Name = "Main Thread";Calculator cal = new Calculator();AddDelegate del = new AddDelegate(cal.Add);IAsyncResult asyncResult = del.BeginInvoke(2, 5, null, null); // 异步调用方法// 做某些其它的事情，模拟需要执行3 秒钟for (int i = 1; i <= 3; i++){Thread.Sleep(TimeSpan.FromSeconds(i));Console.WriteLine("{0}: Client executed {1} second(s).", Thread.CurrentThread.Name, i);}int rtn = del.EndInvoke(asyncResult);Console.WriteLine("Result: {0}\n", rtn);Console.WriteLine("\nPress any key to exit...");Console.ReadLine();} }public class Calculator{public int Add(int x, int y){if (Thread.CurrentThread.IsThreadPoolThread){Thread.CurrentThread.Name = "Pool Thread";}Console.WriteLine("Method invoked!");// 执行某些事情，模拟需要执行2 秒钟for (int i = 1; i <= 2; i++){Thread.Sleep(TimeSpan.FromSeconds(i));Console.WriteLine("{0}: Add executed {1} second(s).", Thread.CurrentThread.Name, i);}Console.WriteLine("Method complete!");return x + y;} } 此时的输出为： // Client application started! Method invoked! Main Thread: Client executed 1 second(s). Pool Thread: Add executed 1 second(s). Main Thread: Client executed 2 second(s). Pool Thread: Add executed 2 second(s). Method complete! Main Thread: Client executed 3 second(s). Result: 7 Press any key to exit... // 现在执行完这段代码只需要3 秒钟时间，两个for 循环所产生的输出交替进行，这也说明了这两段代码并行执行的情况。可以看到Add() 方法是由线程池中的线程在执行， 因为Thread.CurrentThread.IsThreadPoolThread 返回了True，同时我们对该线程命名为了Pool Thread。另外我们可以看到通过EndInvoke()方法得到了返回值。有时候，我们可能会将获得返回值的操作放到另一段代码或者客户端去执行，而不是向上面那样直接写在BeginInvoke()的后面。比如说我们在Program 中新建一个方法GetReturn()，此时可以通过AsyncResult 的AsyncDelegate 获得del 委托对象，然后再在其上调用EndInvoke()方法，这也说明了AsyncResult 可以唯一的获取到与它相关的调用了的方法（或者也可以理解成委托对象）。所以上面获取返回值的代码也可以改写成这样： private static int GetReturn(IAsyncResult asyncResult){AsyncResult result = (AsyncResult)asyncResult;AddDelegate del = (AddDelegate)result.AsyncDelegate;int rtn = del.EndInvoke(asyncResult);return rtn;} 然后再将int rtn = del.EndInvoke(asyncResult);语句改为int rtn = GetReturn(asyncResult);。注意上面IAsyncResult 要转换为实际的类型AsyncResult 才能访问AsyncDelegate 属性，因为它没有包含在IAsyncResult 接口的定义中。 BeginInvoke 的另外两个参数分别是AsyncCallback 和Object 类型，其中AsyncCallback 是一个委托类型，它用于方法的回调，即是说当异步方法执行完毕时自动进行调用的方法。它的定义为： // public delegate void AsyncCallback(IAsyncResult ar); // Object 类型用于传递任何你想要的数值，它可以通过IAsyncResult 的AsyncState 属性获得。下面我们将获取方法返回值、打印返回值的操作放到了OnAddComplete()回调方法中： using System.Threading;using System;using System.Runtime.Remoting.Messaging;public delegate int AddDelegate(int x, int y);class Program9{static void Main(string[] args){Console.WriteLine("Client application started!\n");Thread.CurrentThread.Name = "Main Thread";Calculator cal = new Calculator();AddDelegate del = new AddDelegate(cal.Add);string data = "Any data you want to pass.";AsyncCallback callBack = new AsyncCallback(OnAddComplete);del.BeginInvoke(2, 5, callBack, data); // 异步调用方法// 做某些其它的事情，模拟需要执行3 秒钟for (int i = 1; i <= 3; i++){Thread.Sleep(TimeSpan.FromSeconds(i));Console.WriteLine("{0}: Client executed {1} second(s).", Thread.CurrentThread.Name, i);}Console.WriteLine("\nPress any key to exit...");Console.ReadLine();}static void OnAddComplete(IAsyncResult asyncResult){AsyncResult result = (AsyncResult)asyncResult;AddDelegate del = (AddDelegate)result.AsyncDelegate;string data = (string)asyncResult.AsyncState;int rtn = del.EndInvoke(asyncResult);Console.WriteLine("{0}: Result, {1}; Data: {2}\n", Thread.CurrentThread.Name, rtn, data);} }public class Calculator{public int Add(int x, int y){if (Thread.CurrentThread.IsThreadPoolThread){Thread.CurrentThread.Name = "Pool Thread";}Console.WriteLine("Method invoked!");// 执行某些事情，模拟需要执行2 秒钟for (int i = 1; i <= 2; i++){Thread.Sleep(TimeSpan.FromSeconds(i));Console.WriteLine("{0}: Add executed {1} second(s).", Thread.CurrentThread.Name, i);}Console.WriteLine("Method complete!");return x + y;} } 它产生的输出为： Client application started! Method invoked! Main Thread: Client executed 1 second(s). Pool Thread: Add executed 1 second(s). Main Thread: Client executed 2 second(s). Pool Thread: Add executed 2 second(s). Method complete! Pool Thread: Result, 7; Data: Any data you want to pass. Main Thread: Client executed 3 second(s). Press any key to exit... 这里有几个值得注意的地方： 1、我们在调用BeginInvoke()后不再需要保存IAysncResult 了，因为AysncCallback 委托将该对象定义在了回调方法的参数列表中； 2、我们在OnAddComplete()方法中获得了调用BeginInvoke()时最后一个参数传递的值，字符串“Any data you want to pass”； 3、执行回调方法的线程并非客户端线程Main Thread，而是来自线程池中的线程Pool Thread。另外如前面所说，在调用EndInvoke()时有可能会抛出异常，所以在应该将它放到try/catch 块中，这里就不再示范了。 1.8 总结 我们详细地讨论了C中的委托和事件，包括什么是委托、为什么要使用委托、事件的由来、.NET Framework 中的委托和事件、委托中方法异常和超时的处理、委托与异步编程、委托和事件对Observer 设计模式的意义。拥有了本章的知识，相信你以后遇到委托和事件时，将不会再有所畏惧。 本篇文章为转载内容。原文链接：https://blog.csdn.net/beyonddeg/article/details/53528482。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...一步探讨面向对象设计模式中的“策略模式”与该技巧的结合运用。策略模式是一种行为设计模式，允许在运行时根据上下文动态选择算法实现。通过使用成员函数指针，我们可以在策略模式中更灵活地定义和切换不同的执行策略。 近期，许多现代C++库（如Boost和C++11标准库）中的部分组件就巧妙运用了成员函数指针作为策略模式的一部分。例如，在处理事件驱动编程时，可以将不同类型的对象及其处理事件的非静态成员函数注册为事件处理器，当特定事件触发时，通过调用存储的成员函数指针达到执行对应策略的目的。 此外，随着C++17引入的std::invoke函数以及可变参数模板，对成员函数指针的使用变得更加简洁和安全。std::invoke能够正确处理各种可调用对象（包括成员函数指针），并确保无论何种类型都能以统一的方式进行调用。 总的来说，掌握将非静态成员函数作为参数传递给函数指针的技术，不仅有助于解决实际编程问题，更能为设计复杂系统、实现灵活多变的行为策略提供强大支持。不断跟进C++新特性的学习和实践，是提升面向对象设计和编码能力的关键所在。

...及异步编程，尤其是在处理事件监听和定时器时，闭包的作用尤为关键。 近期，随着WebAssembly技术的不断发展与成熟，Lua因其轻量级和高性能的特性，被越来越多地应用于WebAssembly环境中的脚本编写。在这种场景下，闭包的灵活运用有助于开发者更高效地管理内存资源和实现复杂的状态逻辑。 同时，针对闭包可能导致的内存泄漏问题，社区内有持续的研究与探讨。例如，LuaJIT项目通过改进垃圾回收机制，有效缓解了因闭包产生的内存泄露风险。而一些先进的编程实践和模式，如函数式编程风格下的纯函数使用，可以在一定程度上避免无意识地创建长期持有外部状态的闭包。 此外，对于深入理解和掌握闭包这一概念，推荐读者进一步研读《Programming in Lua》一书，书中对Lua语言特性和闭包原理有着详尽而系统的阐述，并提供了大量实用示例以供学习参考。通过理论与实践相结合的方式，开发者能够更好地驾驭闭包这一强大工具，从而提升代码质量和程序性能。

...ZooKeeper的事件处理机制：一次深入浅出的探索之旅 1. 引言 当我们谈论分布式系统时，ZooKeeper这个名字总会自然而然地浮现在我们的眼前。ZooKeeper这款神奇的小工具，它可是个分布式、开源的协调服务大拿，在管理集群、维护配置、提供命名服务这些重要环节里，都起着不可或缺的关键作用。而其强大的事件处理机制，则是支撑其高效稳定运行的核心要素之一。大家好，这次咱们要一起深入地“摸透”ZooKeeper这家伙的事件处理机制，我保证会让你像看故事一样轻松理解。不仅如此，咱还会结合实实在在的代码实例，让你亲手感受这个机制究竟有多大的魔力，准备好了吗？咱们这就开始探索之旅吧！ 2. ZooKeeper事件概述 在ZooKeeper的世界里，客户端与服务器之间的交互主要通过一系列事件触发和响应来完成。这些事件涵盖了节点创建、删除、更新以及监听器的注册和触发等场景。比方说，当你在ZooKeeper里头新建了一个小节点，或者数据悄咪咪发生了变化的时候，ZooKeeper这个家伙可机灵了，它会立马告诉那些提前报名登记过、时刻关注这些变动的客户端们。 3. ZooKeeper事件类型 ZooKeeper定义了一系列丰富的事件类型： - CREATED：当节点被创建时触发。 - DELETED：当节点被删除时触发。 - CHANGED：当节点数据发生改变时触发。 - CHILDREN_CHANGED：当子节点列表发生变更时触发。 java import org.apache.zookeeper.Watcher.Event.EventType; public enum EventType { Created, Deleted, Changed, ChildEvent } 4. ZooKeeper监听器注册与使用 为了处理这些事件，我们需要在客户端实现一个Watcher接口，并将其注册到感兴趣的ZooKeeper节点上。 java import org.apache.zookeeper.Watcher; public interface Watcher { void process(WatchedEvent event); } 下面是一个简单的监听器实现示例： java public class MyWatcher implements Watcher { @Override public void process(WatchedEvent event) { if (event.getType() == EventType.NodeCreated) { System.out.println("Node created: " + event.getPath()); } else if (event.getType() == EventType.NodeDeleted) { System.out.println("Node deleted: " + event.getPath()); } // 其他事件类型的处理... } } 然后，在ZooKeeper客户端初始化后，我们可以这样注册监听器： java ZooKeeper zookeeper = new ZooKeeper("localhost:2181", 3000, new MyWatcher()); zookeeper.exists("/myNode", true); // 注册对/myNode节点的监听 在这个例子中，当"/myNode"节点的状态发生变化时，MyWatcher类中的process方法就会被调用，从而执行相应的事件处理逻辑。 5. 事件的一次性特性 值得一提的是，ZooKeeper的监听器是一次性的——即事件一旦触发，该监听器就会被移除。如果想持续监听某个节点的变化，需要在process方法中重新注册监听器。 java @Override public void process(WatchedEvent event) { // 处理事件逻辑... // 重新注册监听器 zookeeper.exists(event.getPath(), this); } 6. 结语 ZooKeeper的事件处理机制无疑为其在分布式环境中的强大功能奠定了基石。它使得各个组件可以实时感知到状态变化，并据此做出快速响应。这次咱们深入研究了ZooKeeper这家伙的事件处理机制，不仅摸清了它背后的玄机，还亲眼见识到了在实际开发中它是如何被玩转、如何展现其灵活性的。这种机制的设计理念，对于我们理解和构建更复杂、更健壮的分布式系统具有深远的启示意义。希望各位在阅读这篇内容的时候，能真真切切地体验到这个机制的独门秘籍，然后把它活学活用，让这股独特魅力在未来你们的实际项目操作中大放异彩。

...后，我们发现实时数据处理与流计算领域的发展日新月异。最近，Apache Flink 1.14版本也推出了对事件时间和 watermark 的改进，进一步强化了其在复杂事件处理和乱序数据管理上的能力。该版本优化了watermark生成逻辑，并引入了更为灵活的event time策略配置，使得开发者能够更好地应对不同业务场景下的延迟数据挑战。 另外，随着物联网、金融交易、社交网络等领域的快速发展，实时数据的价值日益凸显，对流处理系统提出了更高要求。例如，阿里巴巴在其2021年双十一活动中，就运用了升级版的实时计算引擎，结合事件时间驱动的数据一致性保障机制，确保了数十亿级别交易数据的实时统计分析准确性。 同时，学术界也在不断探索和完善实时数据处理理论框架，如加州大学伯克利分校AMPLab团队提出的“Lambda架构”，以及斯坦福大学DINOSAUR项目中的“Kappa架构”，都在尝试以不同的方式整合Processing Time和Event Time，旨在构建更高效、更健壮的实时数据处理解决方案。 因此，在实际应用Spark Structured Streaming进行实时数据处理时，关注行业动态和技术前沿，对比研究其他流处理框架的时间模型处理方式，将有助于我们更好地适应快速变化的数据环境，设计出更加符合业务需求的数据处理策略。

...严格的类型检查、错误处理机制以及更高效的运行时性能。在本文案例中，开发者使用ActionScript 3.0编写代码来实现客户端与服务器端的数据交互逻辑，包括连接建立、事件监听、函数调用等操作。 NetConnection , 在ActionScript 3.0中，NetConnection类主要用于客户端与服务器之间的网络连接，支持点对点或客户端-服务器模式的实时双向通信。在文中实例中，NetConnection对象被用来建立客户端到Flash Media Server (FMS) 的连接，从而允许客户端调用服务器端脚本函数并接收返回结果。 Flash Media Server (FMS) , Flash Media Server 是Adobe公司提供的流媒体服务器软件，能够高效地传输和管理音频、视频及数据流内容。在这个例子中，FMS作为后台服务端，负责响应来自Flash客户端（通过NetConnection建立连接）的请求，执行相应的服务器端脚本，并将处理后的数据回传给客户端。 Responder , 在ActionScript 3.0的NetConnection API中，Responder对象用于处理从服务器端调用返回的结果或者错误信息。在文章示例中，当客户端调用服务器端方法时，会创建一个Responder实例，并定义了success和failed两个回调函数，分别处理成功获取数据和调用失败的情况。

一、引言 Flink CEP（复杂事件处理）是Apache Flink的一个功能强大的模块，它可以让用户在大数据环境中进行实时分析。处理复杂的事件，其实就像是在无尽的数据洪流里淘宝，目标是要挖出那些真正有价值的、有意义的信息，这种方式可以说是一种高级的数据处理技术。 二、应用场景 1. 实时监控系统 在实时监控系统中，我们需要从大量的实时数据流中获取有价值的信息，例如设备故障、异常行为等。Flink CEP可以帮助我们实时地发现这些事件，并及时采取措施。 java StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream> stream = env.addSource(new DataStreamSource<>(new FileInputFormat<>("file:///path/to/input/file"))).map(new MapFunction, Tuple2>() { @Override public Tuple2 map(Tuple2 value) throws Exception { // 将字符串转为整数 return new Tuple2<>(value.f0, Integer.parseInt(value.f1)); } }); Pattern, Tuple2> pattern = Pattern., Tuple2>begin("start") .where(new FilterFunction>() { @Override public boolean filter(Tuple2 value) throws Exception { // 判断是否满足条件 return value.f1 > 10; } }) .next("middle") .where(new FilterFunction>() { @Override public boolean filter(Tuple2 value) throws Exception { // 判断是否满足条件 return value.f1 > 20; } }) .followedByAny("end"); DataStream>> results = pattern.grep(stream); results.print(); env.execute("Flink CEP Example"); 这段代码中，我们首先定义了一个事件模式，该模式包含三个事件，分别名为“start”、“middle”和“end”。然后，我们就在这串输入数据流里头“抓”这个模式，一旦逮到匹配的，就把它全都给打印出来。拿这个例子来说吧，我们想象一下，“start”就像是你按下开关启动一台机器的那一刻；“middle”呢，就好比这台机器正在呼呼运转，忙得不可开交的时候；而“end”呢，就是指你再次关掉开关，让设备安静地停止工作的那个时刻。设备一旦启动运转起来，要是过了10秒这家伙还在持续运行没停下来的话，那咱们就可以把它判定为“不正常行为”啦。 2. 实时推荐系统 在实时推荐系统中，我们需要根据用户的实时行为数据生成个性化的推荐结果。Flink CEP可以帮助我们实现实时的推荐计算。 python from pyflink.datastream import StreamExecutionEnvironment, DataStream, ValueStateDescriptor from pyflink.table import DataTypes, TableConfig, StreamTableEnvironment, Schema, \ BatchTableEnvironment, TableSchema, Field, StreamTableApi env = StreamExecutionEnvironment.get_execution_environment() t_config = TableConfig() t_env = StreamTableEnvironment.create(env, t_config) source = ... t_env.connect JDBC("url", "username", "password") \ .with_schema(Schema.new_builder() \ .field("user_id", DataTypes.STRING()) \ .field("product_id", DataTypes.STRING()) \ .field("timestamp", DataTypes.TIMESTAMP(3)) \ .build()) \ .with_name("stream_table") \ .create_temporary_view() pattern = Pattern( from_elements("order", DataTypes.STRING()), OneOrMore( PatternUnion( Pattern.of_type(DataTypes.STRING()).equalTo("purchase"), Pattern.of_type(DataTypes.STRING()).equalTo("click"))), to_elements("session")) result = pattern.apply(t_env.scan("stream_table")) result.select("order_user_id").print_to_file("/tmp/output") env.execute("CEP example") 在这段代码中，我们首先创建了一个表环境，并从JDBC连接读取了一张表。然后，我们定义了一个事件模式，该模式包含了两个事件：“order”和“session”。最后，我们使用这个模式来筛选表中的数据，并将结果保存到文件中。这个例子呢，我们把“order”想象成一次买买买的行动，而“session”呢，就相当于一个会话的开启或者结束，就像你走进商店开始挑选商品到结账离开的整个过程。当用户连续两次剁手买东西，或者接连点啊点的，我们就会觉得这位朋友可真是活跃得不得了，然后我们就把他的用户ID美滋滋地记到文件里去。 3. 实时告警系统 在实时告警系统中，我们需要在接收到实时数据后立即发送告警。Flink CEP可以帮助我们实现实时的告

...ndra表结构？ 在处理海量时序数据的场景下，Apache Cassandra是一个非常出色的选择。它的分布式架构以及对大数据读写操作的高度优化，使其成为存储和查询时间序列数据的理想平台。不过，有效地利用Cassandra的前提是精心设计数据模型。本文将带你手把手地深入挖掘，如何为时间序列数据量身打造Cassandra的表结构设计。咱会借助实例代码和亲身实战经验，像揭开宝藏地图那样揭示其中的设计秘诀，让你明明白白、实实在在地掌握这门技艺。 1. 理解时间序列数据特点 时间序列数据是指按时间顺序记录的一系列数据点，每个数据点通常与一个特定的时间戳相关联。这类数据在咱们日常生活中可不少见，比如物联网（IoT）、监控系统、金融交易还有日志分析这些领域，都离不开它。它的特点就是会随着时间的推移，像滚雪球一样越积越多。而在查询的时候，人们最关心的通常就是最近产生的那些新鲜热辣的数据，或者根据特定时间段进行汇总统计的信息。 2. 设计原则 （1）分区键选择 在Cassandra中，分区键对于高效查询至关重要。当你在处理时间序列数据时，一个很接地气的做法就是拿时间来做分区的一部分。比如说，你可以把年、月、日、小时这些信息拼接起来，弄成一个复合型的分区键。这样一来，同一时间段的数据就会乖乖地呆在同一个分区里，这样咱们就能轻松高效地一次性读取到这一整段时期的数据了，明白吧？ cql CREATE TABLE sensor_data ( sensor_id uuid, event_time timestamp, data text, PRIMARY KEY ((sensor_id, date_of(event_time)), event_time) ) WITH CLUSTERING ORDER BY (event_time DESC); 这里date_of(event_time)是对事件时间进行提取日期部分的操作，形成复合分区键，便于按天或更粗粒度进行分区。 （2）排序列簇与查询路径 使用CLUSTERING ORDER BY定义排序列簇，按照时间戳降序排列，确保最新数据能快速获取。 （3）限制行大小与集合使用 尽管Cassandra支持集合类型，但对于时间序列数据，应避免在一个集合内存放大量数据，以免读取性能受到影响。由于集合不会分页，如果需要存储连续的时序数据点，最好让每一行只包含单个数据点。 （4）宽行与稀疏索引 采用“宽行”策略，即每行代表一段时间窗口内的多个数据点属性，而不是每条数据一个行。这有助于减少跨分区查询，提高查询效率。同时呢，对于那些跟时间没关系的筛选条件，我们可以琢磨着用一下稀疏索引。不过得注意啦，这里有个“度”的把握，就是索引虽然能让查询速度嗖嗖提升，但同时也会让写入数据时的开销变大。所以嘞，咱们得在这两者之间找个最佳平衡点。 3. 示例设计 物联网传感器数据存储 假设我们有一个物联网项目，需要存储来自不同传感器的实时测量值： cql CREATE TABLE sensor_readings ( sensor_id uuid, reading_time timestamp, temperature float, humidity int, pressure double, PRIMARY KEY ((sensor_id, reading_time)) ) WITH CLUSTERING ORDER BY (reading_time DESC); 这个表结构中，sensor_id和reading_time共同组成复合分区键，每个传感器在某一时刻的温度、湿度和压力读数都存放在一行里。 4. 总结与思考 设计Cassandra时间序列数据表的关键在于理解数据访问模式并结合Cassandra的特性和局限性。选对分区键这招儿，就像给海量数据找个宽敞的储藏室，让它们能分散开来存放和快速找到；而把列簇整得井井有条，那就相当于帮我们轻松摸到最新鲜的数据，一抓一个准儿。再配上精心设计的宽行结构，加上恰到好处的索引策略，甭管查询需求怎么变花样，都能妥妥地满足你。 当然，具体实践时还需要根据业务的具体情况进行调整和优化，例如预测未来的数据增长规模、评估查询性能瓶颈以及是否需要进一步的数据压缩等措施。总的来说，用Cassandra搭建时间序列数据模型不是个一劳永逸的事儿，它更像是一个持久的观察、深度思考和反复调整优化的过程。只有这样，我们才能真正把Cassandra处理海量时序数据的洪荒之力给释放出来。

...引言 Mahout与Flink的完美融合 在数据科学的领域里，Mahout和Flink都是不可或缺的利器。Mahout，一个开源的机器学习库，以其强大的算法库而闻名，尤其在推荐系统、聚类分析和协同过滤等领域有着广泛的应用。哎呀，你知道Flink这个家伙吗？这家伙可是个了不得的工具！它就像个超级英雄一样，专门负责处理那些海量的数据流，而且速度超快，延迟超低，简直就像闪电侠附体似的。用它来实时分析数据，那简直就是小菜一碟，分分钟搞定！当这两者相遇，一场数据处理的革命便悄然发生。 二、Mahout的Flink接口 功能概述 Mahout的Flink接口提供了丰富的功能，旨在将Mahout的机器学习能力与Flink的实时计算能力相结合，为用户提供更高效、更灵活的数据分析工具。以下是几个核心功能： 1. 实时推荐系统构建 通过Flink流处理特性，Mahout可以实时处理用户行为数据，快速生成个性化推荐，提升用户体验。 2. 大规模聚类分析 利用Flink的并行处理能力，Mahout能对大量数据进行高效聚类，帮助发现数据中的模式和结构。 3. 在线协同过滤 Flink接口允许Mahout实现在线协同过滤算法，实时更新用户偏好，提高推荐的准确性和时效性。 4. 数据流上的机器学习 Mahout的Flink接口支持在数据流上执行机器学习任务，如实时异常检测、预测模型更新等。 三、代码示例 构建实时推荐系统 为了更好地理解Mahout的Flink接口如何工作，下面我们将构建一个简单的实时推荐系统。哎呀，这个玩意儿啊，它能根据你过去咋用它的样子，比如你点过啥，买过啥，然后啊，它就能实时给你推东西。就像是个超级贴心的朋友，老记着你的喜好，时不时给你点惊喜！ java import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class RealtimeRecommendationSystem { public static void main(String[] args) throws Exception { // 创建流处理环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 假设我们有一个实时事件流，包含用户ID和商品ID DataStream> eventStream = env.fromElements( Tuple2.of("user1", "itemA"), Tuple2.of("user2", "itemB"), Tuple2.of("user1", "itemC") ); // 使用Mahout的协同过滤算法进行实时推荐 DataStream> recommendations = eventStream.map(new MapFunction, Tuple2>() { @Override public Tuple2 map(Tuple2 value) { // 这里只是一个示例，实际应用中需要调用具体的协同过滤算法 return new Tuple2<>(value.f0, "recommendedItem"); } }); // 打印输出 recommendations.print(); // 执行任务 env.execute("Realtime Recommendation System"); } } 四、结论 开启数据驱动的未来 通过整合Mahout的机器学习能力和Flink的实时计算能力，开发者能够构建出响应迅速、高效精准的数据分析系统。无论是实时推荐、大规模聚类还是在线协同过滤，这些功能都为数据分析带来了新的可能。哎呀，随着科技这玩意儿越变越厉害，咱们能见到的新鲜事儿也是一波接一波。就像是魔法一样，数据这东西，现在能帮咱们推动业务发展，搞出不少新花样，让咱们的生意越来越红火，创意源源不断。简直就像开了挂一样！

批流一体处理：在Apache Flink中切换between Batch and Streaming modes 批处理和流处理是大数据处理中的两种核心模式，而Apache Flink以其独特的设计理念实现了批与流的一体化处理。本文将深入探讨Flink如何无缝切换并高效执行批处理和流处理任务，并通过丰富的代码示例帮助你理解这一机制。 1. Apache Flink 批流一体的统一计算引擎 （1）Flink的设计哲学 Apache Flink的核心理念是将批视为一种特殊的流——有限流，从而实现了一种基于流处理的架构去同时处理无限流数据和有界数据集。这种设计简直让开发者们乐开了花，从此以后再也不用头疼选择哪种处理模型了。无论是对付那些堆积如山的历史数据，还是实时流动的数据流，都能轻松驾驭，只需要同一套API就能搞定编写工作。这样一来，不仅开发效率噌噌噌地往上飙，连资源利用率也得到了前所未有的提升，真可谓是一举两得的超级福利！ （2）批流一体的实现原理 在Flink中，所有的数据都被视作数据流，即便是静态的批数据，也被看作是无界流的一个切片。这就意味着，批处理的任务其实可以理解为流处理的一个小弟，只需要在数据源那里设定一个特定的边界条件，就一切搞定了。这么做的优点就在于，开发者能够用一个统一的编程套路，来应对各种不同的应用场景，轻轻松松实现批处理和流处理之间的无缝切换。就像是你有了一个万能工具箱，甭管是组装家具还是修理电器，都能游刃有余地应对，让批处理和流处理这两种模式切换起来就像换扳手一样自然流畅。 2. 切换批处理与流处理模式的实战演示 （1）定义DataStream API java import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class BatchToStreamingExample { public static void main(String[] args) throws Exception { // 创建流处理环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 假设这是批处理数据源（实际上Flink也支持批处理数据源） DataStream text = env.fromElements("Hello", "World", "Flink", "is", "awesome"); // 流处理操作（映射函数） DataStream mappedStream = text.map(new MapFunction() { @Override public String map(String value) { return value.toUpperCase(); } }); // 在流处理环境中提交作业（这里也可以切换到批处理模式下运行） env.execute("Batch to Streaming Example"); } } （2）从流处理模式切换到批处理模式 上述代码是在流处理环境下运行的，但实际上，只需简单改变数据源，我们就可以轻松地处理批数据。例如，我们可以使用readTextFile方法读取文件作为批数据源： java DataStream text = env.readTextFile("/path/to/batch/data.txt"); 在实际场景中，Flink会根据数据源的特性自动识别并调整内部执行策略，实现批处理模式下的优化执行。 3. 深入探讨批流一体的价值 批处理和流处理模式的无缝切换，不仅简化了编程模型，更使资源调度、状态管理以及故障恢复等底层机制得以统一，极大地提高了系统的稳定性和性能表现。同时呢，这也意味着当业务需求风吹草动时，咱能更灵活地扭动数据处理策略，不用大费周章重构大量代码。说白了，就是“一次编写，到处运行”，真正做到灵活应变，轻松应对各种变化。 总结来说，Apache Flink凭借其批流一体的设计理念和技术实现，让我们在面对复杂多变的大数据应用场景时，拥有了更为强大且高效的武器。无论你的数据是源源不断的实时流，还是静待处理的历史批数据，Flink都能游刃有余地完成使命。这就是批流一体的魅力所在，也是我们深入探索和研究它的价值所在。

一、引言 Flink是一个强大的流处理框架，它可以帮助我们高效地处理海量数据。在用Flink干活儿的时候，咱们免不了会碰到各种幺蛾子，其中最多人吐槽的就是状态存储这茬儿。好嘞，那咱们今天就唠唠嗑，说说这怎么挑个合适的State Backend吧！ 二、什么是State Backend？ 在Flink中，我们经常需要保存一些中间结果或者上下文信息，这就是所谓的状态。而这些状态的存储方式就被称为State Backend。Flink提供了多种不同的State Backend，包括RocksDB、FsState等。 三、选择State Backend的原则 当我们面临选择State Backend的问题时，我们需要遵循以下几个原则： 3.1 稳定性 这是最重要的一个原则。咱们得挑一个超级稳定的State Backend，这样咱的应用才能稳如磐石，不会因为State Backend抽风而突然罢工。 3.2 性能 性能也是一个重要的考虑因素。我们得挑一个超级给力的State Backend，这样一来，咱们的应用运行起来就能溜得飞起，效率杠杠的。 3.3 可扩展性 随着我们的应用规模的扩大，我们需要选择一个可扩展性强的State Backend，这样可以满足我们未来的需求。 四、RocksDB State Backend RocksDB是一种高性能的键值对数据库，它是Google开源的一个项目。Flink提供了一个基于RocksDB的State Backend。 java ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.setStateBackend(new RocksDBStateBackend("/tmp/flink-rocksdb")); 五、FsState State Backend FsState是Flink提供的一个基于文件系统的State Backend。 java ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.setStateBackend(new FsStateBackend("/tmp/flink-fsstate")); 六、总结 选择合适的State Backend是一项非常重要的任务。咱们应该根据自身的实际需求和所处的环境条件，来挑个最适合的State Backend，就像选衣服要根据身材和天气一样，得找准那个最合拍的“款”。同时呢，咱们也得留意这么个事儿，就是各种State Backend各有各的好和不足。要想做出最合适的决定，就得先把这些家伙的脾性摸个透彻明白才行。 以上就是我对于如何选择合适的State Backend的一些理解和看法，希望能够对你有所帮助。如果你有任何问题或者想法，欢迎留言讨论。 七、尾声 Flink是一个强大且灵活的流处理框架，但是它的复杂性也给我们带来了一些挑战。我们需要不断地学习和探索，才能更好地利用它。在挑State Backend的时候，咱们得根据自身的实际情况和需求，像个精明的买家那样，选出最对胃口、最适合的那个选项。

...如何在Apache Flink中定义一个数据源——Source。Flink，这个强大的流处理工具，可厉害了！它让我们能够随心所欲地定义各种数据源。比如说，文件系统里存的那些数据、数据库里躺着的各种记录，甚至是从网络上飞来飞去的信息，全都可以被咱们轻松纳入囊中，没有啥太大的限制！ 二、什么是Source？ 在Flink中，Source是一个用于产生数据并将其转换为适合流处理的形式的组件。它是一个特殊的Operator，其输入是0或多个其他Operators的输出，而其输出则是进一步处理的数据流。 三、如何在Flink中定义一个数据源？ 定义一个Source非常简单，只需要遵循以下几个步骤： 第一步：选择你的数据源 首先，你需要确定你要从哪里获取数据。这完全可能是个文件夹、数据库什么的，也可能是网络呀，或者实时传感器这类玩意儿，反正只要是能提供数据的来源，都行！ 第二步：创建Source类 接下来，你需要创建一个Source类来表示你的数据源。这个类需要继承自org.apache.flink.api.common.functions.SourceFunction接口，并实现run方法。 例如，如果你的数据源是从一个文件系统中读取的文本文件，你可以创建一个这样的Source类： java public class MySource implements SourceFunction { private boolean isRunning = true; @Override public void run(SourceContext ctx) throws Exception { File file = new File("/path/to/my/file.txt"); try (BufferedReader reader = new BufferedReader(new FileReader(file))) { String line; while ((line = reader.readLine()) != null && isRunning) { ctx.collect(line); } } } @Override public void cancel() { isRunning = false; } } 在这个例子中，我们的Source类MySource会从指定路径的文件中读取每一行并发送给下游的Operators进行处理。 第三步：注册Source到StreamGraph 最后，你需要将你的Source注册到一个StreamGraph中。你可以通过调用StreamExecutionEnvironment.addSource方法来完成这个操作。 例如： java StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream dataStream = env.addSource(new MySource()); 四、总结 以上就是我们在Flink中定义一个数据源的基本步骤。当然啦，实际情况可能还会复杂不少，比如说你可能得同时对付多个数据来源，或者先给数据做个“美容”（预处理）啥的。不过，只要你把基础的概念和技术都玩得溜溜的，这些挑战对你来说就都不是事儿，你可以灵活应对，轻松解决。 五、结语 我希望这篇文章能帮助你更好地理解和使用Flink中的Source。如果你有任何问题或者想要分享你的经验，欢迎留言讨论。让我们一起学习和进步！ 六、附录 参考资料 1. Apache Flink官方文档 https://ci.apache.org/projects/flink/flink-docs-latest/ 2. Java 8 API文档 https://docs.oracle.com/javase/8/docs/api/ 3. Stream Processing with Flink: A Hands-on Guide by Kostas Tsichlas and Thomas Hotham (Packt Publishing, 2017).

Flink on YARN部署与资源管理策略：一次深度探索之旅 1. 引言 Apache Flink，作为一款开源的流处理和批处理大数据框架，以其高效、灵活的特点深受开发者喜爱。实际上，很多工程师都非常关心一个核心问题，那就是如何在拥有大量机器的集群环境下，巧妙地借助YARN（这个资源协商小能手）来把Flink任务部署得妥妥当当，同时又能把各种资源调配管理得井井有条。本文将带领大家深入探讨Flink on YARN的部署方式，并通过实例代码揭示其背后的资源配置策略。 2. Flink on YARN部署初探 2.1 部署原理 当我们选择在YARN上运行Flink时，实质上是将Flink作为一个YARN应用来部署。YARN就像个大管家，它会专门给Flink搭建一个叫做Application Master的“指挥部”。这个“AM”呢，就负责向YARN这位资源大佬申请干活所需要的“粮草物资”，然后根据Flink作业的具体需求，派遣出一队队TaskManager“小分队”去执行实际的计算任务。 bash 启动Flink作业在YARN上的Application ./bin/flink run -m yarn-cluster -yn 2 -ys 1024 -yjm 1024 -ytm 2048 /path/to/your/job.jar 上述命令中，-yn指定了TaskManager的数量，-ys和-yjm分别设置了每个容器的内存大小和Application Master的内存大小，而-ytm则定义了每个TaskManager的内存大小。 2.2 配置详解 - -m yarn-cluster 表示在YARN集群模式下运行Flink作业。 - -yn 参数用于指定TaskManager的数量，可以根据实际需求调整以适应不同的并发负载。 - -ys、-yjm 和 -ytm 则是针对YARN资源的细致调控，确保Flink作业能在合理利用集群资源的同时，避免因资源不足而导致的性能瓶颈或OOM问题。 3. 资源管理策略揭秘 3.1 动态资源分配 Flink on YARN支持动态资源分配，即在作业执行过程中，根据当前负载情况自动调整TaskManager的数量。这种策略极大地提高了资源利用率，特别是在应对实时变化的工作负载时表现突出。 3.2 Slot分配机制 在Flink内部，资源被抽象为Slots，每个TaskManager包含一定数量的Slot，用来执行并行任务。在YARN这个大环境下，我们能够灵活掌控每个TaskManager能同时处理的任务量。具体来说，就是可以根据TaskManager内存的大小，还有咱们预先设置的slots数量，来精准调整每个TaskManager的承载能力，让它恰到好处地执行多个任务并发运行。 例如，在flink-conf.yaml中设置： yaml taskmanager.numberOfTaskSlots: 4 这意味着每个TaskManager将提供4个slot，也就是说，理论上它可以同时执行4个并发任务。 3.3 自定义资源请求 对于特殊的场景，如GPU密集型或者高CPU消耗的作业，我们还可以自定义资源请求，向YARN申请特定类型的资源。不过这需要YARN环境本身支持异构资源调度。 4. 结语 关于Flink on YARN的思考与讨论 理解并掌握Flink on YARN的部署与资源管理策略，无疑能够帮助我们在面对复杂的大数据应用场景时更加游刃有余。不过同时也要留意，实际操作时咱们得充分照顾到业务本身的特性，还有集群当前的资源状况，像玩拼图一样灵活运用这些策略。不断去微调、优化资源分配的方式，确保Flink能在YARN集群里火力全开，达到最佳效能状态。在这个过程中，我们会不断地挠头琢磨、动手尝试、努力改进，这恰恰就是大数据技术最吸引人的地方——它就像一座满是挑战的山峰，但每当你攀登上去，就会发现一片片全新的风景，充满着无限的可能性和惊喜。 通过以上的阐述和示例，希望你对Flink on YARN有了更深的理解，并在未来的工作中能更好地驾驭这一强大的工具。记住，技术的魅力在于实践，不妨现在就动手试一试吧！

...和加速Web开发的预定义代码库。在本文中，JQuery即是一款流行的JavaScript框架，它提供了一系列便捷的方法来操作HTML文档对象模型(DOM)，处理事件、执行动画效果以及实现Ajax交互等任务。 HTML DOM结构 , DOM（Document Object Model）是W3C组织制定的一种标准，它将HTML或XML文档表示为树形结构，允许程序和脚本动态更新、添加、删除和修改页面内容及结构。在文章中，通过JQuery可以高效地对HTML DOM结构进行遍历、查找、修改等操作。 事件委托 , 事件委托是JavaScript中一种处理事件绑定的技术，尤其适用于动态生成的DOM元素。在JQuery中，使用.on()方法实现事件委托，即将一个事件处理器绑定到父元素上，当该父元素的后代元素触发指定事件时，会由父元素的事件处理器统一处理。这样做的好处在于，不仅节省内存开销，还能确保即使未来新增的子元素也能正确响应事件。在本文示例中，通过事件委托技术，我们能够确保所有动态生成的.item元素都能正确响应click事件。

...核心竞争力。深入研究事件委托、异步编程等核心概念，以及如何利用CSS3的新特性（如伪类选择器、自定义属性等）来丰富界面交互效果，将帮助开发者在实际项目中游刃有余地应对各种需求变化，无论是使用jQuery还是其他前端解决方案。

...者大量采用箭头函数来处理事件和状态变更，此时由于箭头函数不会创建自己的this上下文，因此避免了this指向问题带来的困扰，同时也简化了代码逻辑。然而，这也要求开发者更精准地掌握闭包、作用域链等相关知识，以应对无this特性的函数调用情况。 此外，对于Class组件中的方法，新版JavaScript允许直接在类体内定义实例方法并利用箭头函数自动绑定this，如myMethod = () => {...}，从而省去了手动bind(this)的操作，提升了代码的简洁性和可读性。 值得注意的是，在异步编程领域，Promise和async/await的广泛应用也带来了this绑定的新挑战。尤其是在回调函数或者异步任务中，需要结合EventEmitter、class实例或bind等方式确保this的正确引用。 总之，随着JavaScript语言的持续演进以及各种前端框架的更新迭代，理解和运用好this关键字显得更为关键。同时，关注社区最新实践和规范，不断适应新的编程范式，也是每个前端开发者必备的技能之一。

...DOM等特性，在表单处理与文件上传方面展现出更高效和灵活的优势。例如，Vue.js通过其v-model指令和自定义组件可以轻松实现表单数据绑定，并结合axios等库进行异步文件提交。 然而，JQuery仍然是许多遗留项目及对轻量化有较高需求场景下的可靠选择。值得注意的是，尽管原生JQuery不直接支持现代浏览器的fetch API，但开发者可以通过引入第三方插件（如jQuery-File-Upload）或者采用原生JavaScript配合FormData对象，实现更为先进的多文件上传、进度跟踪等功能。 最近，Bootstrap团队在最新版本中依然保持了对JQuery的兼容性，这进一步证明了即使在现代化前端生态下，JQuery仍然具有广泛的用户基础和实用价值。同时，对于初学者而言，理解并掌握JQuery的工作原理及其在表单提交方面的应用，有助于深入理解DOM操作和事件处理机制，为学习更复杂的前端框架打下坚实的基础。 另外，随着WebAssembly等新技术的发展，未来表单文件提交的方式可能会有更多创新突破。例如，利用WebAssembly可以在浏览器端实现高性能的数据预处理和加密，从而提高文件上传的安全性和效率。因此，无论是坚守经典的JQuery方案，还是拥抱新兴的前端框架和技术，持续关注和探索这一领域的发展动态都是至关重要的。

...文中，jQuery被定义为一款优秀的JavaScript库，它提供了丰富的函数和方法，可以简化HTML文档操作、事件处理以及动画效果实现等任务。 兼容性问题 , 在Web开发领域，兼容性问题是指由于不同浏览器对HTML、CSS和JavaScript等Web标准支持程度的差异，导致网页或应用在某些浏览器上无法正常显示或功能缺失的现象。文中提到，jQuery在IE8及以下版本的浏览器中存在兼容性问题，需要通过特定扩展来解决这些问题，确保其功能可以在多种浏览器环境下稳定运行。 Polyfill（此处虽未直接出现“polyfill”一词，但String.prototype.trim的自定义实现可视作一种polyfill） , Polyfill是一种编程技术，用于在不支持特定功能的老旧浏览器中提供该功能的模拟实现。例如，文章中提到的为IE8添加对String.prototype.trim方法的支持，即创建了一个polyfill，使得即使在不支持trim原生方法的IE8浏览器中，也能使用相同的语法进行字符串去空格操作。 AJAX（$.ajaxSetup提及） , AJAX全称Asynchronous JavaScript and XML，是一种创建动态网页应用的技术，允许在不刷新整个页面的情况下与服务器交换数据并更新部分网页内容。在jQuery中，$.ajaxSetup是一个全局配置方法，用来设置所有后续$.ajax()请求的默认选项。在本文情境下，为了规避IE浏览器中的缓存问题，建议设置$.ajaxSetup()的cache属性为false，以保证每次AJAX请求都能获取最新的服务器响应。 动画效果（animate方法提及） , 在Web开发中，动画效果通常指元素在网页上的动态变化，如大小、位置、透明度等属性的变化过程。jQuery提供的.animate()方法就是用来帮助开发者更方便地创建动画效果。文中指出，在IE8浏览器下，jQuery的.animate()方法对opacity属性的支持存在问题，需要通过修改此方法的实现来保证透明度动画能在IE8浏览器中正常工作。

...化HTML文档遍历、事件处理、动画以及Ajax交互等功能的开发。它封装了JavaScript语言的核心功能，提供了一种更加简洁、易于理解且具有高度可读性的语法结构，使得开发者能够更高效地操作网页DOM元素、处理事件和执行异步请求。 DOM（Document Object Model）操作 , 在Web开发中，DOM是浏览器对HTML或XML文档的一种内部表示方式，允许开发者通过JavaScript等脚本语言动态地访问、修改和创建网页内容与结构。在本文提到的jQuery教程中，DOM操作是指使用jQuery提供的方法来选取、修改、添加或删除HTML元素，例如更改元素样式、内容，插入新的元素，或者隐藏显示已有元素等行为。 插件 , 在jQuery框架下，插件是一种扩展jQuery功能的独立模块。开发者可以基于jQuery核心库开发出特定用途的功能组件，这些组件可以在项目中直接引入并调用，以实现诸如表单验证、图片轮播、拖拽效果等各种复杂功能。教程中的插件章节介绍了如何正确安装和使用jQuery社区或其他开发者编写的第三方插件，帮助开发者增强网站的用户体验和交互性。

...in Apache Flink 一、引言（≈150字） 在Apache Flink的流处理与批处理应用开发中，我们常常会遇到一个名为“TypeInformationException”的异常，其中一种典型错误提示就是：“Missing type information for generic type parameter”。这种异常主要源于Flink对Java泛型类型的识别和处理机制。这篇文章呢，咱们要来个深度挖掘，把这个异常现象背后的小秘密给揪出来，还会配上些实实在在的代码例子，一起唠唠怎么才能真正地防止和搞定这个问题。 二、理解TypeInformationException（≈250字） 在Flink的世界里，TypeInformation扮演着至关重要的角色。它包含了数据类型的所有必要信息，如类型是否可null、是否基本类型、是否有字段以及字段的类型等。对于使用了泛型的数据类型，Flink需要获取到具体的类型参数信息以便正确处理。当Flink无法自动推断出泛型的具体类型时，就会抛出"Missing type information for generic type parameter"的异常。 三、案例分析（≈300字 + 代码示例 ≈ 150字） 假设我们在Flink作业中定义了一个泛型类Event，并尝试将其作为DataStream的元素类型： java public class Event { private T payload; // ... getters and setters } DataStream> stream = env.addSource(new FlinkSource>()); 运行上述代码时，Flink就无法确定T的具体类型，从而引发"TypeInformationException"。因为?通配符表示任何类型，Flink无法从Event推导出确切的TypeInformation。 为了解决这个问题，我们需要显式地提供TypeInformation： java TypeInformation> stringTypeInfo = TypeInformation.of(new TypeHint>() {}); DataStream> stream = env.addSource(new FlinkSource<>(stringTypeInfo)); 四、深入解决方案（≈250字 + 代码示例 ≈ 150字） 另一种更为通用的方法是使用TypeInformation.of()或TypeExtractor.createTypeInfo()方法，结合TypeHint或自定义的TypeInformation子类来明确指定泛型参数的类型： java // 使用TypeHint方式 TypeInformation> integerTypeInfo = TypeInformation.of(new TypeHint>() {}); DataStream> integerStream = env.addSource(new FlinkSource<>(integerTypeInfo)); // 或者使用TypeExtractor方式 TypeInformation> doubleTypeInfo = TypeExtractor.getForClass(Event.class) .forGenericTypes(Double.class); DataStream> doubleStream = env.addSource(new FlinkSource<>(doubleTypeInfo)); 五、思考与总结（≈200字） 面对“Missing type information for generic type parameter”这类异常，我们需要理解其背后的原理：Flink为了确保数据处理的正确性和效率，必须清楚每种数据类型的细节。所以，说到泛型这事儿，开发者们最好积极拥抱Flink的类型系统，明确地提供各类类型信息，别藏着掖着~此外，在设计数据模型时，尽可能避免过度复杂的泛型结构也能降低此类问题的发生概率。记住了啊，编程不只是敲出能跑起来的代码那么简单，更重要的是要深入理解并完全掌握系统的底层运作机制。这样一来，无论遇到什么难题挑战，都能像庖丁解牛那样游刃有余地应对处理。

...化了Web开发人员在处理HTML文档、操作DOM元素、处理事件以及执行Ajax交互等任务时的工作。通过提供简洁易读的API和丰富的插件生态系统，JQuery使得开发者能够快速实现诸如动画效果、表单验证、网页内容筛选等功能，从而提高开发效率并增强用户体验。 JavaScript库 , JavaScript库是一组预先编写的、可复用的JavaScript代码集合，旨在为开发者提供便利，简化常见的编程任务，例如DOM操作、Ajax请求、事件处理、动画制作等。在本文中，JQuery就是一个用于简化网页开发的JavaScript库，它封装了许多复杂的JavaScript功能，使得开发者可以使用更简洁、易于理解的语法来完成复杂任务。 DOM遍历（文中提及的段落遍历） , DOM遍历是指在HTML文档对象模型（Document Object Model, DOM）中查找、访问或操作每一个节点的过程。在本文上下文中，通过JQuery的each()方法遍历ID为“content”的div元素下的所有段落（p标签），逐个检查其文本内容是否包含用户在搜索框中输入的关键字，进而实现搜索文字变色的功能。 keyup事件 , keyup事件是JavaScript中的一个DOM事件，当用户释放键盘上的任意键后触发。在本文示例中，我们为搜索框绑定了keyup事件监听器，这样每当用户在搜索框中输入或修改关键词后松开按键，就会触发相应的JavaScript函数，实时更新页面内匹配关键词的文字高亮状态。 CSS样式（文中提及的highlight类） , CSS（层叠样式表）是一种样式表语言，用于描述HTML或XML（包括如SVG、MathML等各种XML方言）文档的呈现。在文章中提到的.highlight类样式，就是在CSS中定义的一种样式规则，用来给匹配到搜索关键词的文本添加背景颜色（黄色），从而实现高亮显示的效果。