Reactive Extensions いいよ Reactive Extensions。
Reactive Extensions を Java に移植したのが reactive4java。
.NET じゃないので Linq やラムダ式は使えない(いやスクリプトとしてなら使える)けど、並列処理を直列に書けるヨロコビは味わえます。

Android でも使えます。
Android の端末の方位を通知する処理を書いてみました(仕事で必要だったので)。

SensorFunctions.java

import hu.akarnokd.reactive4java.base.Func1;
import hu.akarnokd.reactive4java.base.Functions;
import hu.akarnokd.reactive4java.reactive.Observable;
import hu.akarnokd.reactive4java.reactive.Observer;
import hu.akarnokd.reactive4java.reactive.Reactive;

public final class SensorFunctions {

	/**
	 * 端末の確度を Observable として取得する
	 * 
	 * ※端末は縦向き前提です。横向きには対応していません。
	 * port from sato-c / sensorcheck https://github.com/sato-c/sensorcheck
	 */
	static public Observable<Float> getRotationAsObservable(final Context context, final int rate) {
		return Reactive.createWithCloseable(new Func1<Observer<? super Float>, Closeable>() {

			@Override
			public Closeable invoke(final Observer<? super Float> observer) {
	    		final AtomicBoolean _stop = new AtomicBoolean(false); // 停止フラグ
				
				final SensorManager accelMan = (SensorManager)context.getSystemService(Context.SENSOR_SERVICE);
				final SensorManager magMan   = (SensorManager)context.getSystemService(Context.SENSOR_SERVICE);

		    	final SensorEventListener listener = new SensorEventListener() {
		    		final static private int _MATRIX_SIZE = 16;
		    		final static private int _XYZ_AXIS = 3;
		    		private boolean _sensorReady = false;
		    		private float[] _R = new float[_MATRIX_SIZE];
		    		private float[] _I = new float[_MATRIX_SIZE];
		    		private float[] _magnetic = null;
		    		private float[] _accel = null;
		    		private float[] _orient = new float[_XYZ_AXIS];
		    		
					/*
					 * 	http://developer.android.com/reference/android/hardware/SensorEvent.html#values
					 */
					@Override
					public void onSensorChanged(SensorEvent event) {
						// 停止されていたら読み飛ばす
						if (_stop.get()) {
							return;
						}
						
						switch ( event.sensor.getType() ) {
						case	Sensor.TYPE_ACCELEROMETER:
							_accel = event.values.clone();
							break;
						case	Sensor.TYPE_MAGNETIC_FIELD:
							_sensorReady = true;
							_magnetic = event.values.clone();
							break;
						}
						
						if ( _sensorReady && _magnetic != null && _accel != null ) {
							_sensorReady = false;
							
					        SensorManager.getRotationMatrix(_R, _I, _accel, _magnetic);
					        SensorManager.getOrientation(_R, _orient);
	
					        if ( _orient[0] < 0 ) {
					        	_orient[0] = (float) (Math.toDegrees(_orient[0]) + 360.0f);
					        } else {
					        	_orient[0] = (float) (Math.toDegrees(_orient[0]));
					        }
					        
					        // 発火
					        observer.next(_orient[0]);
						}
					}
					
					@Override
					public void onAccuracyChanged(Sensor sensor, int accuracy) {
					}
				};
		    	
				Sensor accelSensor = getSensorOnce(accelMan, Sensor.TYPE_ACCELEROMETER);
				if (accelSensor == null) {
					observer.error(new InvalidParameterException("Sensor.TYPE_ACCELEROMETER not found."));
					return Functions.EMPTY_CLOSEABLE;
				}

				Sensor magSensor = getSensorOnce(accelMan, Sensor.TYPE_MAGNETIC_FIELD);
				if (magSensor == null) {
					observer.error(new InvalidParameterException("Sensor.TYPE_MAGNETIC_FIELD not found."));
					return Functions.EMPTY_CLOSEABLE;
				}
				
				accelMan.registerListener(listener, accelSensor, rate);
				magMan.registerListener(listener, magSensor, rate);
				
				return new Closeable() {
					@Override
					public void close() throws IOException {
						// 受信はすぐに止まらないのでフラグで読み飛ばす
						if (_stop.get()) {
							return;
						}
						_stop.set(true);
						
						magMan.unregisterListener(listener);
						accelMan.unregisterListener(listener);
					}
				};
			}
			
			private Sensor getSensorOnce(SensorManager sensorMan, int sensorType) {
		    	List<Sensor> sensors = sensorMan.getSensorList(sensorType);
		    	if (sensors.size() == 0) {
		    		return null;
		    	}
		    	
		    	return sensors.get(0);
			}
		});
	}
}

使うほうはこんな感じ。

Closeable closer = 
	SensorFunctions.getRotationAsObservable(
		context, SensorManager.SENSOR_DELAY_NORMAL))
.register(new Observer<Float>() { // Listen 開始
	@Override
	public void next(Float value) {
		// 角度が変更されると呼ばれる
		Log.d(TAG, "next() called. - " + value);
	}

	@Override
	public void finish() {
		// 終了(Closeable.Close) すると呼ばれる
		Log.d(TAG, "finish() called.");
	}

	@Override
	public void error(Throwable err) {
		// なんかエラー
		Log.e(TAG, "error() called.", err);
	}
});

Thread.sleep(3000); // 3秒待って

closer.Close(); // Listen 終了

.register() すると、角度が変わる度に、 next() が呼ばれ、角度が通知されます。
（処理自体は非同期で行われるので、上の例だと Sleep を挟まないとすぐに Listen 停止してしまいます。）

これ実際に実行すると、すごい勢いで next が呼ばれて大変なことになります。

これじゃデータ多すぎるよぉ・・・そんな時も Reactive Extensions は便利。

こうします。

Closeable closer = 
ObservableBuilder.from( // メソッドチェーンできるおまじない
	SensorFunctions.getRotationAsObservable(
		context, SensorManager.SENSOR_DELAY_NORMAL)))
.sample(1, TimeUnit.SECONDS) // 1秒置きの最新値にフィルタ
.register(new Observer<Float>() { // Listen 開始
	@Override
	public void next(Float value) {
		// 角度が変更されると呼ばれる
		Log.d(TAG, "next() called. - " + value);
	}

	@Override
	public void finish() {
		// 終了(Closeable.Close) すると呼ばれる
		Log.d(TAG, "finish() called.");
	}

	@Override
	public void error(Throwable err) {
		// なんかエラー
		Log.e(TAG, "error() called.", err);
	}
});

Thread.sleep(3000); // 3秒待って

closer.Close(); // Listen 終了

すばらしい！ .sample(1, TimeUnit.SECONDS) を挟むだけで、「１秒間隔での取得」に置き換えることができます。

他にもいろいろ便利な機能がある reactive4java ですが、とりまこんなもんで。

##追記 9.27
実はこのプログラム、Cold でしたー。というわけで、Hot についての記事を書きました。

Cold を Hot にできる。そう、Publish ならね。 - Qiita

##参考

Reactive Extensionsで非同期処理を簡単に - Rx の一番わかりやすいスライドだと思います！
neue cc - Rx
Reactive Extensions - かずきのBlog@Hatena
reactive4java
TYPE_ORIENTATIONを使わずに方位角を取得 - プログラマー'sペイジ
sato-c / sensorcheck - 方位を取得する処理を移植させて頂きました