Broadcast  广播
Broadcast 可以将数据广播到每个节点上，这样节点需要使用该数据是无需发送请求，大大提升了效率。

Broadcast 提供了接口，具体的方法在子类中实现，如子类TorrentBroadcast。
源码清单和我的理解注释
ListenerBus源码中第一行，创建了一个线程安全的ArrayList——CopyOnWriteArrayList，之后添加、者删除事件等操作都在这个线程安全的ArrayList中执行
/**
   * 标记广播变量是否有效的标志
   */
  @volatile private var _isValid = true

/** 先判断广播变量是否有效再获取值 */
  def value: T = {
    assertValid()
    getValue()
  }

删除执行器excutor中的广播副本
/**
   * Asynchronously delete cached copies of this broadcast on the executors.
   * If the broadcast is used after this is called, it will need to be re-sent to each executor.
   */
  def unpersist() {
    unpersist(blocking = false)
  }

  /**
   * Delete cached copies of this broadcast on the executors. If the broadcast is used after
   * this is called, it will need to be re-sent to each executor.
   * @param blocking Whether to block until unpersisting has completed
   */
  def unpersist(blocking: Boolean) {
    assertValid()
    doUnpersist(blocking)
  }

摧毁广播变量
 /**
   * Destroy all data and metadata related to this broadcast variable. Use this with caution;
   * once a broadcast variable has been destroyed, it cannot be used again.
   * This method blocks until destroy has completed
   * 摧毁与此广播变量相关的所有数据和元数据，一旦摧毁便无法再使用它
   * 这个函数将被阻塞，直到摧毁工作完成
   */
  def destroy() {
    destroy(blocking = true)
  }

  /**
   * Destroy all data and metadata related to this broadcast variable. Use this with caution;
   * once a broadcast variable has been destroyed, it cannot be used again.
   * @param blocking Whether to block until destroy has completed
   * 摧毁与此广播变量相关的所有数据和元数据，一旦摧毁便无法再使用它
   * 参数blocking可以选择是否在摧毁完成前阻止摧毁工作
   */
  private[spark] def destroy(blocking: Boolean) {
    assertValid()
    _isValid = false
    _destroySite = Utils.getCallSite().shortForm
    logInfo("Destroying %s (from %s)".format(toString, _destroySite))
    doDestroy(blocking)
  }

使用打开关闭数据网络，触发BroadCast

1、新建BroadCastTest工程；

2、新建MyBroadCast

   在OnReceive（）增加一句显示代码 Toast.maketext（）.show();

3、在MainActivity中定义两个变量；

   private IntentFilter intentFilter;启动intent，带action；

   private  MyBroadCast myBroadCast;

  intentFilter=new IntentFilter();

  intentFilter.action("android.net.conn.CONNECTION_CHANGE");

  myBroadCast=new MyBroadCast();

  regesterreceiver(myBroadCast,intentFilter);//注册自定义的BroadCast，带一个Action；

 

4、OnDestroy（）

    unregesterreceiver(myBroadCast);

 

5、在AndroidMainifest.xml中填写：

     <use-permission android :name="android.permission.ACCESS_NETWORK_STATE"/>

在Android中，Broadcast是一种广泛运用的在应用程序之间传输信息的机制。而BroadcastReceiver是对发送出来的 Broadcast进行过滤接受并响应的一类组件。

下面将详细的阐述如何发送Broadcast和使用BroadcastReceiver过滤接收的过程：

　　首先在需要发送信息的地方，把要发送的信息和用于过滤的信息(如Action、Category)装入一个Intent对象，然后通过调用 sendOrderBroadcast()或sendStickyBroadcast()方法，把 Intent对象以广播方式发送出去。

　　当Intent发送以后，所有已经注册的BroadcastReceiver会检查注册时的IntentFilter是否与发送的Intent相匹配，若匹配则就会调用BroadcastReceiver的onReceive()方法。所以当我们定义一个BroadcastReceiver的时候，都需要实现onReceive()方法。

　　注册BroadcastReceiver有两种方式:

　　静态注册：在AndroidManifest.xml中用标签生命注册，并在标签内用标签设置过滤器。

　　<receiver android:name="myRecevice">    //继承BroadcastReceiver，重写onReceiver方法

　　　　<intent-filter>    

　　　　　　<action android:name="com.dragon.net"></action> //使用过滤器，接收指定action广播

　　　   </intent-filter>

　　</receiver> 

　　动态注册：

　　IntentFilter intentFilter = new IntentFilter();

　　intentFilter.addAction(String);   //为BroadcastReceiver指定action，使之用于接收同action的广播

      registerReceiver(BroadcastReceiver,intentFilter);

　　一般：在onStart中注册，onStop中取消unregisterReceiver

　　指定广播目标Action：Intent intent = new Intent(actionString);

　　并且可通过Intent携带消息 :intent.putExtra("msg", "hi,我通过广播发送消息了");

　　发送广播消息：Context.sendBroadcast(intent )

 

其中在动态注册中可将BroadcastReceiver的继承类进行封装，添加构造函数和BroadcastReceiver注册

 

代码


import android.app.Notification;
import android.app.NotificationManager;
import android.app.PendingIntent;
import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;

public class BroadcastReceiverHelper extends BroadcastReceiver {

    NotificationManager mn=null;
    Notification notification=null;
    Context ct=null;
    BroadcastReceiverHelper receiver;

public BroadcastReceiverHelper(Context c){
        ct=c;
        receiver=this;
    }

//注册
    public void registerAction(String action){
        IntentFilter filter=new IntentFilter();
        filter.addAction(action);
        ct.registerReceiver(receiver, filter);
    }

    @Override
public void onReceive(Context context, Intent intent) {
// TODO Auto-generated method stub
        String msg=intent.getStringExtra("msg");
int id=intent.getIntExtra("who", 0);
if(intent.getAction().equals("com.cbin.sendMsg")){
                mn=(NotificationManager)context.getSystemService(Context.NOTIFICATION_SERVICE);
                notification=new Notification(R.drawable.icon, id+"发送广播", System.currentTimeMillis());
                Intent it = new Intent(context,Main.class);
                PendingIntent contentIntent=PendingIntent.getActivity(context,
0, it, 0);
                notification.setLatestEventInfo(context, 
"msg", msg, contentIntent);
                mn.notify(0, notification);
            }
    }
}




 

然后再Activity中声明BroadcastReceiver的扩展对象，在onStart中注册，onStop中卸载

 

代码


    
BroadcastReceiverHelper  rhelper;
@Override
public void onStart(){
//注册广播接收器
        rhelper=new BroadcastReceiverHelper(this);
        rhelper.registerAction("com.cbin.sendMsg");
super.onStart();
    }

    @Override
public void onStop(){
//取消广播接收器
        unregisterReceiver(rhelper);
super.onStop();
    }




 总结：

1、broadcastreceiver 处理时间不得超过10秒，否则出现ANR错误。

2、broadcastreceiver中onreceive执行完后，该receiver对象就被销毁了，下次使用会重新创建新的receiver对象。

3、静态注册的receiver在其程序启动后就常驻系统，即使该程序退出ondestroy后（或重启手机后），receiver仍能接收广播。除非卸载该程序。

    动态注册的receiver在程序退出后就不再接收广播。

Broadcast是分布式的数据共享，由BroadcastManager负责管理其创建或销毁。Broadcast一般用于处理共享的配置文件、通用Dataset、常用数据结构
1.1、创建
通过SparkContext.broadcast广播一个Broadcast, 实际调用的是SparkEnv的BroadManager来创建
  /**
   * Broadcast a read-only variable to the cluster, returning a
   * [[org.apache.spark.broadcast.Broadcast]] object for reading it in distributed functions.
   * The variable will be sent to each cluster only once.
   */
  def broadcast[T: ClassTag](value: T): Broadcast[T] = {
    assertNotStopped()
    require(!classOf[RDD[_]].isAssignableFrom(classTag[T].runtimeClass),
      "Can not directly broadcast RDDs; instead, call collect() and broadcast the result.")
    //使用SparkEnv.broadcastManager创建Broadcast
    val bc = env.broadcastManager.newBroadcast[T](value, isLocal)
    val callSite = getCallSite
    logInfo("Created broadcast " + bc.id + " from " + callSite.shortForm)
    cleaner.foreach(_.registerBroadcastForCleanup(bc))
    bc
  }

1.1.1、在SparkEnv中创建BroadcastManager，
// 此处只是声明， 只有调用initialize， 才会生效
val broadcastManager = new BroadcastManager(isDriver, conf, securityManager)

1.1.2、initialize
// Called by SparkContext or Executor before using Broadcast
  private def initialize() {
    synchronized {
      if (!initialized) {
        broadcastFactory = new TorrentBroadcastFactory
        broadcastFactory.initialize(isDriver, conf, securityManager)
        initialized = true
      }
    }
  }

1.1.3、 BoradcastManager操作BradCast实际是代理BroadcastFactory, 此处使用工厂模式
  def stop() {
    broadcastFactory.stop()
  }

  private val nextBroadcastId = new AtomicLong(0)

  def newBroadcast[T: ClassTag](value_ : T, isLocal: Boolean): Broadcast[T] = {
    broadcastFactory.newBroadcast[T](value_, isLocal, nextBroadcastId.getAndIncrement())
  }

  def unbroadcast(id: Long, removeFromDriver: Boolean, blocking: Boolean) {
    broadcastFactory.unbroadcast(id, removeFromDriver, blocking)
  }