前言
在微服務架構的系統(tǒng)中，我們通常會使用輕量級的消息代理來構建一個共用的消息主題讓系統(tǒng)中所有微服務實例都能連接上來诀紊，由于該主題中產(chǎn)生的消息會被所有實例監(jiān)聽和消費槐臀，所以我們稱它為消息總線。在總線上的各個實例都可以方便地廣播一些需要讓其他連接在該主題上的實例都知道的消息县忌，例如配置信息的變更或者其他一些管理操作等僧凰。
由于消息總線在微服務架構系統(tǒng)的廣泛使用探颈，所以它同配置中心一樣，幾乎是微服務架構中的必備組件训措。spring cloud作為微服務架構綜合性的解決方案伪节，對此自然也有自己的實現(xiàn)，這就是spring cloud bus绩鸣。通過spring cloud bus怀大，可以非常容易的搭建起消息總線，同時實現(xiàn)了一些消息總線中的常用功能呀闻，比如配合spring cloud config實現(xiàn)微服務應用配置信息的動態(tài)更新等化借。
消息代理
消息代理（message broker）是一種消息驗證，傳輸捡多，路由的架構模式蓖康。它在應用程序之間起到通信并最小化應用之間的依賴的作用，使得應用程序可以高效地解耦通信過程垒手。消息代理是一個中間件產(chǎn)品蒜焊，它的核心是一個消息的路由程序，用來實現(xiàn)接收和分發(fā)消息科贬，并根據(jù)設定好的消息處理流來轉(zhuǎn)發(fā)給正確的應用泳梆。它包括獨立的通信和消息傳遞協(xié)議，能夠?qū)崿F(xiàn)組織內(nèi)部和組織間的網(wǎng)絡通信榜掌。設計代理的目的就是為了能夠從應用程序中傳入消息优妙，并執(zhí)行一些特別的操作，下面這些是企業(yè)應用中，我們經(jīng)常使用消息代理的場景：
- 將消息路由到一個或多個目的地。
- 消息轉(zhuǎn)化為其他的表現(xiàn)方式。
- 執(zhí)行消息的聚集，消息的分解，并將結果發(fā)送到它們的目的地朋贬，然后重新組合響應返回給消息用戶航罗。
- 調(diào)用web服務來檢索數(shù)據(jù)。
- 響應事件或錯誤抄罕。
- 使用發(fā)布-訂閱模式來提供內(nèi)容和基于主題的消息路由。
目前已經(jīng)有非常多的開源產(chǎn)品可以供大家使用于颖，比如：
- activemq
- kafka
- rabbitmq
- rocketmq
- ...
當前版本的spring cloud bus僅支持兩款中間件產(chǎn)品：rabbitmq和kafka呆贿。
rabbitmq實現(xiàn)消息總線
rabbitmq是實現(xiàn)了高級消息隊列協(xié)議（AMQP）的開源消息代理軟件，也稱為面向消息的中間件森渐。Rabbitmq服務是高性能做入，可伸縮性而聞名的Erlang語言編寫而成的，其集群和故障轉(zhuǎn)移是構建在開放電信平臺框架的同衣。
AMQP是Advanced Message Queuing Protocol的簡稱竟块，它是一個面向消息中間件的開發(fā)式標準應用層協(xié)議，它定義了以下這些特性：
- 消息方向
- 消息隊列
- 消息路由（包括點到點和發(fā)布-訂閱模式）
- 可靠性
- 安全性
AMQP要求消息的提供者和客戶端接收者的行為要實現(xiàn)對不同的供應商可以用相同的方式（比如SMTP耐齐，HTTP浪秘，F(xiàn)TP等）進行互相操作。在以往的中間件標準中埠况，主要還是建立在api級別的耸携，比如jms，集中于通過不同的中間件實現(xiàn)來建立標準化的程序間的互操作性辕翰，而不是在多個中間件產(chǎn)品間實現(xiàn)互操作性夺衍。
AMQP與JMS不同，JMS定義了一個API和一組消息收發(fā)必須要實現(xiàn)的行為喜命，而AMQP是一個線路級協(xié)議沟沙。線路級協(xié)議描述的是通過網(wǎng)絡發(fā)送的數(shù)據(jù)傳輸格式。因此壁榕，任何符合該數(shù)據(jù)格式的消息發(fā)送和接收工具都能互相兼容和進行操作矛紫，這樣就能輕易實現(xiàn)跨技術平臺的架構方案。
RabbitMQ以AMQP協(xié)議實現(xiàn)护桦，所以它可以支持多種操作系統(tǒng)含衔，多種編程語言，幾乎可以覆蓋所有主流的企業(yè)級技術平臺二庵。在微服務架構消息中間件的選型中贪染，它是一個非常適合且優(yōu)秀的選擇。因此催享，在spring cloud bus中包含了對rabbit的自動化默認配置杭隙。
基本概念
介紹一些Rabbitmq的基本概念，
- Broker：可以理解成消息隊列服務器的實體因妙，它是一個中間件應用痰憎，負責接收消息生產(chǎn)者的消息票髓，然后將消息發(fā)送到消息接收者或者其他的Broker。
- Exchange：消息交換機铣耘，是消息第一個到達的地方洽沟，消息通過它指定的路由規(guī)則，分發(fā)到不同的消息隊列中去蜗细。
- Queue：消息隊列裆操，消息通過發(fā)發(fā)送和路由之后最終到達的地方，到達Queue的消息即進入邏輯上等待消費的狀態(tài)炉媒。每個消息都會被發(fā)送到一個或多個隊列踪区。
- Binding：綁定，它的作用就是把Exchange和Queue按照路由規(guī)則綁定起來吊骤，也就是Exchange和Queue之間的虛擬連接缎岗。
- Routing Key：路由關鍵字，Exchange根據(jù)這個關鍵字進行消息投遞白粉。
- Virtual host：虛擬主機传泊，它是對Broker的虛擬劃分，將消費者蜗元，生產(chǎn)者和它們的依賴的AMQP相關結構進行隔離或渤，一般都是為了安全考慮。比如奕扣，我們可以在一個Broker中設置多個虛擬主機薪鹦，對不同用戶進行權限的分離。
- Connection：連接惯豆，代表生產(chǎn)者池磁，消費者，Broker之間進行通信的物理網(wǎng)絡楷兽。
- Channel：消息通道地熄，用于連接生產(chǎn)者和消費者的邏輯結構。在客戶端的每個連接里芯杀，可建立多個Channel端考，每個Channel代表一個會話任務，通過Channel可以隔離同一個連接中的不同交互內(nèi)容揭厚。
- Producer：消息生產(chǎn)者却特，制造消息并發(fā)送消息的程序。
- Consumer：消息消費者筛圆，接收消息并處理消息的程序裂明。
消息投遞到隊列的整個過程大致如下：
1.客戶端連接到消息隊列服務器，打開一個Channel太援。
2.客戶端聲明一個Exchange闽晦，并設置相關屬性扳碍。
3.客戶端聲明一個Queue，并設置相關屬性仙蛉。
4.客戶端使用Routing Key笋敞，在Exchange和Queue之間建立好綁定關系。
5.客戶端投遞消息到Exchange捅儒。
- Exchange接收到消息后液样，根據(jù)消息的key和已經(jīng)設置的Binding，進行消息路由巧还，將消息投遞到一個或多個Queue里。
Exchange也有幾種類型坊秸。
1.Direct交互機：完全根據(jù)Key進行投遞麸祷。比如，綁定時設置了Routing Key為abc褒搔，那么客戶端提交的消息阶牍，只有設置了key為Routing Key的才會被投遞到隊列。
2.Topic交互機：對Key進行模式匹配后進行投遞星瘾，可以使用符號#匹配一個或多個詞走孽，符號*匹配正好一個詞。比如,abc.#匹配abc.def.ghi, abc.*只匹配abc.def.
3.Fanout交互機：不需要任何Key琳状，它采用廣播的模式磕瓷，一個消息進來時，投遞到與該交互機綁定的所有隊列念逞。
Rabbitmq支持消息持久化困食，也就是將數(shù)據(jù)寫在磁盤上。為了數(shù)據(jù)安全考慮翎承，大多數(shù)情況下都會選擇持久化硕盹。消息隊列持久化包括三個部分：
- Exchange持久化，在聲明時指定durable >=1.
- Queue持久化叨咖，在聲明時指定durable => 1.
- 消息持久化瘩例，在投遞時指定delivery_mode => 2(1是非持久化）。
如果Exchange和Queue都是持久化甸各，那么它們之間的Binding也是持久化的垛贤。如果Exchange和Queue兩者之間有一個是持久化的，一個是非持久化的痴晦，就不允許建立綁定南吮。
安裝
快速入門
在springboot中整合Rabbitmq是一個非常容易的事情，
- 新建一個
spring boot工程誊酌，命名為springboot-rabbitmq - 在pom文件中引入依賴部凑，其中
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.4.5.RELEASE</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
</dependencies>
- 在
application.properties中配置關于Rabbitmq的連接和用戶信息露乏，
spring:
application:
name: springboot-rabbitmq
rabbitmq:
host:
port: 5672
username:
password:
- 創(chuàng)建生產(chǎn)者
Sender。通過注入AmqpTemplate接口的實例來實現(xiàn)消息的發(fā)送涂邀，AmqpTemplate接口定義了一套針對AMQP協(xié)議的基礎操作瘟仿，在spring boot中會根據(jù)配置來注入其具體的實現(xiàn)。
我們發(fā)送一字符串到zhihao.miao.order隊列中比勉，
@Component
public class Sender {
@Autowired
private AmqpTemplate amqpTemplate;
public void send(){
String context = "hello "+ LocalDateTime.now().toString();
System.out.println("Sender: "+context);
this.amqpTemplate.convertAndSend("zhihao.miao.order",context);
}
}
- 創(chuàng)建消息消費者
Receiver劳较。通過@RabbitListener注解定義該類對指定隊列的監(jiān)聽，并用@RabbitHandler注解來指定對消息的處理方法（不同的消息格式浩聋，@RabbitHandler配置的方法的入?yún)⒕筒挥霉畚希J是byte[] 類型）。所以衣洁，該消費者實現(xiàn)了對zhihao.miao.order隊列的消費墓捻，消費操作作為輸出消息的字符串內(nèi)容。
@Component
@RabbitListener(queues = "zhihao.miao.order")
public class Receiver {
@RabbitHandler
public void process(String hello){
System.out.println("Receiver: "+hello);
}
}
- 創(chuàng)建
RabbitMQ的配置類RabbitConfig坊夫，用來配置隊列砖第，交換機，路由等高級信息环凿。這里我們只配置隊列梧兼，已完成一個基本的生產(chǎn)消費過程。
這一步相當于自動創(chuàng)建的過程智听，如果在控制臺上已經(jīng)創(chuàng)建了該隊列羽杰，此步驟可以省略。
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
public Queue queue(){
return new Queue("zhihao.miao.order");
}
}
- 創(chuàng)建啟動主類
@SpringBootApplication
public class RabbitMQApplication {
public static void main(String[] args) {
SpringApplication.run(RabbitMQApplication.class,args);
}
}
- 創(chuàng)建單元測試類瞭稼，用來調(diào)用消息生產(chǎn)
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = RabbitMQApplication.class)
public class RabbitMQApplicationTest {
@Autowired
private Sender sender;
@Test
public void setSender() throws Exception{
sender.send();
}
}
-
啟動應用主類忽洛，在控制臺看到創(chuàng)建了一個連接rabbitmq的連接
springboot整合rabbitmq啟動創(chuàng)建連接
查看控制面板，查看連接信息
-
運行單元測試類环肘，發(fā)送消息
控制臺顯示發(fā)送了消息
整合spring cloud bus
定義了四個項目欲虚，config-server-eureka（spring cloud config server服務），eureka-server（eureka 服務）悔雹，order-service（訂單服務复哆，也是spring cloud config 客戶端），user-service（用戶服務腌零，也是spring cloud config的客戶端）梯找，在git遠程倉庫中定義了二個項目，分別是user-service-config和order-service-config項目益涧。
- 對其進行改造锈锤，修改pom文件，在
user-service和order-service中增加
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>
- 在
user-service和order-service中的配置文件中增加關于Rabbitmq的連接和用戶信息
spring:
rabbitmq:
host:
port: 5672
username:
password:
- 啟動
config-server-eureka，在啟動user-service和order-service久免，我們可以在user-service和order-service的控制臺上看到如下的內(nèi)容浅辙，在啟動的時候多了一個/bus/refresh請求，
- 訪問兩個服務的請求
http://192.168.5.4:7070/user/index和http://192.168.5.4:6060/order/index查看配置中的配置內(nèi)容阎姥，比如我order-service-config的生產(chǎn)環(huán)境配置的配置信息是：
spring:
datasource:
username: '{cipher}af9b9ea63ce1c027d78c1c3414b425ad6f0093c20c69ad144eacb5a8b4522e7c'
check:
uri: pro-1.0
上面的spring.datasource.username是zhihao.miao的對稱加密
user-service-config配置的配置信息是：
spring:
datasource:
username: user-pro
check:
uri: pro-2.0
還有就是order-service我啟動了二個服務记舆，6061服務，訪問http://192.168.5.4:6061/order/index內(nèi)容和6060的結果一樣
- 接著修改兩個配置文件的內(nèi)容呼巴，
user-service-config和order-service-config的check.url屬性泽腮，修改如下：
spring:
datasource:
username: '{cipher}af9b9ea63ce1c027d78c1c3414b425ad6f0093c20c69ad144eacb5a8b4522e7c'
check:
uri: pro-2.0
user-service服務的配置文件：
spring:
datasource:
username: user-pro
check:
uri: pro-1.0
訪問對應的頁面，發(fā)現(xiàn)這些配置都沒有生效衣赶，發(fā)送/bus/refresh到order-service和user-service诊赊，order-service服務
curl -X POST http://localhost:6060/bus/refresh
發(fā)現(xiàn)6060的控制臺和6061的控制臺都打印了很多輸出內(nèi)容，
這樣只要請求order-service服務上的一個實例就可以更新order-service服務的所有實例的配置屑埋，依靠消息總線的功能實現(xiàn)豪筝。
user-service服務也是，
curl -X POST http://localhost:7070/bus/refresh
原理分析
整個方案的架構如下圖所示摘能，其中包含了git倉庫，config server以及幾個微服務應用的實例敲街，這些微服務應用的實例中都引入了spring cloud bus团搞，所以它們都連接到了rabbitmq的消息總線上了。
當我們將系統(tǒng)啟動起來之后多艇，圖中的"server A"的三個實例會請求Config Server以獲取配置信息逻恐，config server根據(jù)應用配置規(guī)則從git倉庫中獲取配置信息并返回。
此時峻黍，我們需要修改"server A"的屬性复隆。首先，通過git管理工具去倉庫中修改對應的屬性值姆涩，但是這個修改并不會觸發(fā)"server A"實例的屬性更新挽拂。我們向"server A"的實例3發(fā)送post請求，訪問/bus/refresh接口骨饿。此時亏栈。“server A”的實例1和實例2從總線中獲取到宏赘，并重新從config server中獲取它們的配置信息绒北，從而實現(xiàn)配置信息的動態(tài)更新。
而從git倉庫中配置的修改到發(fā)起/bus/refresh的post請求這一步可以通過git倉庫的web hook來自動觸發(fā)察署。由于所有連接到消息總線上的應用都會接收到更新請求闷游，所以在web hook中就不需要維護所有節(jié)點內(nèi)容進行更新，從而解決通過web hook來逐個進行刷新的問題。（一般不會使用web hook功能）
使用git倉庫的web hook進行消息總線的事件自動觸發(fā)
URL：就是自動刷新的地址
當配置文件進行修改時會自動觸發(fā)刷新事件脐往，導致配置文件刷新休吠。
指定刷新范圍
局部刷新
我們通過向服務實例請求spring cloud bus的/bus/refresh接口，從而觸發(fā)了總線上其他服務實例的/refresh钙勃。但是在一些特殊場景下蛛碌，我們希望可以刷新微服務中某個具體實例的配置。
spring cloud bus對這種場景也有很好的支持辖源，/bus/refresh接口提供了一個destination參數(shù)蔚携，用于指定具體要刷新的應用程序。比如克饶，可以刷新user-service的6061端口的服務/bus/refresh?destination=customers:6061,此時總線上的各個應用實例會根據(jù)destination屬性的值來判斷是否為自己的實例名酝蜒，若符合才進行配置刷新，若不符合則忽略該消息矾湃。
再去修改一下order-service-config的配置內(nèi)容亡脑，執(zhí)行刷新：
curl -X POST http://localhost:6061/bus/refresh?destination=order-service:6061
此時從控制臺上也可以看出，6061的控制臺上有刷新克隆倉庫配置到本地的日志邀跃，而同一個服務的不同實例6060去沒有日志輸出霉咨，再去訪問url請求驗證一下
http://192.168.5.4:6061/order/index配置已經(jīng)改了。
http://192.168.5.4:6060/order/index配置沒有改變拍屑。
默認情況下途戒，ApplicationContext ID是spring.application.name:server.port（也就是上面destination參數(shù)后面的order-service:6061），詳見org.springframework.boot.context.ContextIdApplicationContextInitializer.getApplicationId(ConfigurableEnvironment) 方法僵驰。
destination參數(shù)除了可以定位具體的實例之外喷斋，還可以用來定位具體的服務。定位服務的原理是通過spring的PathMatecher（路徑匹配）來實現(xiàn)的蒜茴，比如/bus/refresh?destination=customers:**,該請求會觸發(fā)customers服務的所有實例進行刷新星爪。
再去修改order-service服務的倉庫order-service-config的pro環(huán)境的配置：
spring:
datasource:
username: '{cipher}af9b9ea63ce1c027d78c1c3414b425ad6f0093c20c69ad144eacb5a8b4522e7c'
check:
uri: pro-3.0
執(zhí)行/bus/refresh刷新，訪問對應的頁面發(fā)現(xiàn)同一個order-service服務的不同實例都配置都是刷新了粉私。
curl -X POST http://localhost:6061/bus/refresh?destination=order-service:**
應用的上下文id必須不一樣
我們上面知道ApplicationContext id是由三部分組成（name顽腾，profile，index）毡鉴，name是spring.application.name,profile當前指定的配置文件崔泵，index是${vcap.application.instance_index:${spring.application.index:${server.port:${PORT:null}}}}組成.
The bus tries to eliminate processing an event twice, once from the original ApplicationEvent and once from the queue. To do this, it checks the sending application context id againts the current application context id. If multiple instances of a service have the same application context id, events will not be processed. Running on a local machine, each service will be on a different port and that will be part of the application context id. Cloud Foundry supplies an index to differentiate. To ensure that the application context id is the unique, set spring.application.index to something unique for each instance of a service. For example, in lattice, set spring.application.index=${INSTANCE_INDEX} in application.properties (or bootstrap.properties if using configserver).
spring cloud bus執(zhí)行一次刷新就能自動刷新一個服務下的不同實例。 為了達到這個目的猪瞬，它會檢查發(fā)送應用程序上下文id是否一樣憎瘸。 如果服務的多個實例具有相同的應用程序上下文id，則一次刷新不能刷新這個服務下的所有實例陈瘦。 我們在本地機器上運行幌甘，每個服務將在不同的端口上，這個時候的ApplicationContext id是不一樣的，因為端口不一致锅风。而如果實際生產(chǎn)中都是一個服務的不同實例是部署到不同的服務器上的酥诽，端口，應用名皱埠，當前配置文件（pro）都是一致的肮帐，這樣就會出現(xiàn)刷新事件的不能傳播。 Cloud Foundry提供區(qū)分的索引來標識一個服務的不同實例的ApplicationContext id是唯一的边器。 為了確保應用程序上下文id是唯一的训枢，請將spring.application.index設置為服務的每個實例唯一的值。 例如忘巧，在application.properties中設置spring.application.index = $ {INSTANCE_INDEX}（如果使用configserver恒界，請設置bootstrap.properties）。
自己沒有去測試砚嘴。當使用下面的架構優(yōu)化后通過訪問configserver的url和destination參數(shù)是不是只需要配置configserver的spring.application.index的不一致即可還是configserver和configclient都要去配置十酣。（如果是通過刷新configclient則肯定要配置configclient的spring.application.index就行了）
參考資料
Addressing all instances of a service
Application Context ID must be unique
架構優(yōu)化
既然spring cloud bus的/bus/refresh接口提供了針對服務和實例進行配置更新的參數(shù)，那么我們的架構也可以相應的做出一些調(diào)整际长。在之前的demo中耸采。服務的配置更新需要通過向具體服務中的某個實例發(fā)送請求，再觸發(fā)對整個服務集群的配置更新工育，雖然能實現(xiàn)功能洋幻，但是這樣的結果是，我們指定的應用實例會不同于集群中的其他應用實例翅娶，這樣會增加集群內(nèi)部的復雜度，不利于將來的運維工作好唯。比如竭沫，需要對服務實例進行遷移，那么我們不得不修改web hook中的配置等骑篙。所以要盡可能地讓服務集群中的各個節(jié)點是對等的蜕提。
我們主要做了下面的改動：
1.在config server中引入了spring cloud bus，將配置服務端也加入到消息總線來靶端。
2./bus/refresh請求不再發(fā)送到具體的服務實例上谎势，而是發(fā)送給config server，并通過destination參數(shù)類指定需要更新配置的服務或?qū)嵗?/p>
通過上面的改動杨名，我們的服務實例不需要再承擔觸發(fā)配置更新的職責脏榆。同時，對于git的觸發(fā)等配置都只需要針對config server即可台谍。從而簡化了集群上的一些維護工作须喂。
進行改造吧，
config-server-eureka服務，
加入依賴：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>
配置：
spring:
rabbitmq:
host:
port: 5672
username:
password:
啟動config-server-eureka服務坞生，然后發(fā)現(xiàn)控制臺上有/bus/refresh端點進行輸出仔役。
就以order-service來進行測試吧，啟動二個實例（6060是己，6061）又兵，首先查看order-service-config的pro配置文件內(nèi)容：
spring:
datasource:
username: '{cipher}af9b9ea63ce1c027d78c1c3414b425ad6f0093c20c69ad144eacb5a8b4522e7c'
check:
uri: pro-2.0
訪問localhost:6060/order/index結果顯示是
username=zhihao.miao,check.uri===pro-2.0。
我們對配置文件進行修改
spring:
datasource:
username: '{cipher}af9b9ea63ce1c027d78c1c3414b425ad6f0093c20c69ad144eacb5a8b4522e7c'
check:
uri: pro-3.0
通過去訪問config-server-eureka服務提供的刷新端點(localhost:9090/bus/refresh)進行配置刷新卒废，
curl -X POST http://localhost:9090/bus/refresh?destination=order-service:**
我們發(fā)現(xiàn)這命令的時候沛厨，config-server-eureka的控制臺輸出
而實際的order-service(6060,6061)服務的控制臺也輸出一些操作，比如說刷新升熊，從遠程倉庫git clone新的代碼配置等等俄烁，說明我們的架構優(yōu)化是成功的。
追蹤總線事件
Bus events (subclasses of RemoteApplicationEvent) can be traced by setting spring.cloud.bus.trace.enabled=true. If you do this then the Spring Boot TraceRepository (if it is present) will show each event sent and all the acks from each service instance. Example (from the /trace endpoint):
通過設置spring.cloud.bus.trace.enabled=true可以追蹤消息總線上的事件傳播级野∫惩溃可以通過每個服務的/trace的端點來追蹤所有事件的發(fā)起和消息回執(zhí)。
在user-service中加入配置蓖柔，如下辰企，
spring:
cloud:
bus:
trace:
enabled: true
然后重啟服務，訪問url
http://localhost:7070/trace
頁面顯示如下：
顯示RefreshRemoteApplicationEvent事件從user-service:7070發(fā)起况鸣，傳播到user-service下的所有節(jié)點牢贸。
參考資料
Tracing Bus Events
傳播自己的事件
略
參考資料
Broadcasting Your Own Events
